From 20ef394318f5e389996b3205e8f2958ed0ca2a65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Magdalena=20Grodzi=C5=84ska?= Date: Sat, 11 Jan 2020 19:18:51 +0100 Subject: Add flags to configure udup server addr --- src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'src/main/java/pl/edu') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java index 58b55da..7f2ee1f 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java @@ -62,7 +62,8 @@ public class Agent { Integer port = Integer.getInteger("UDUPServer.port"); Integer timeout = Integer.getInteger("UDUPServer.timeout"); Integer bufsize = Integer.getInteger("UDUPServer.bufsize"); - UDUPServer server = new UDUPServer(InetAddress.getByName("127.0.0.1"), port, bufsize); + InetAddress serverAddr = InetAddress.getByName(System.getProperty("UDUPServer.hostname")); + UDUPServer server = new UDUPServer(serverAddr, port, bufsize); modules.put(ModuleType.UDP, new UDUP(port, timeout, bufsize, server)); return modules; } -- cgit v1.2.3 From 1df1cbf93448ed7d446e3f9e192f86f7dd99708b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Magdalena=20Grodzi=C5=84ska?= Date: Sat, 11 Jan 2020 19:36:42 +0100 Subject: Split agent to config classes --- .../java/pl/edu/mimuw/cloudatlas/agent/Agent.java | 278 +-------------------- .../pl/edu/mimuw/cloudatlas/agent/AgentConfig.java | 113 +++++++++ .../mimuw/cloudatlas/agent/HierarchyConfig.java | 171 +++++++++++++ 3 files changed, 292 insertions(+), 270 deletions(-) create mode 100644 src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java create mode 100644 src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java (limited to 'src/main/java/pl/edu') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java index 7f2ee1f..690198b 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java @@ -1,281 +1,19 @@ package pl.edu.mimuw.cloudatlas.agent; -import java.net.Inet4Address; -import java.net.InetAddress; -import java.net.SocketException; -import java.net.UnknownHostException; -import java.rmi.registry.LocateRegistry; -import java.rmi.registry.Registry; -import java.rmi.server.UnicastRemoteObject; -import java.util.List; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Map; -import java.util.Random; -import java.util.Set; -import java.util.concurrent.CompletableFuture; -import java.util.function.Supplier; - -import pl.edu.mimuw.cloudatlas.agent.NewApiImplementation; -import pl.edu.mimuw.cloudatlas.agent.messages.*; -import pl.edu.mimuw.cloudatlas.agent.modules.*; -import pl.edu.mimuw.cloudatlas.agent.modules.Module; -import pl.edu.mimuw.cloudatlas.api.Api; -import pl.edu.mimuw.cloudatlas.interpreter.Main; -import pl.edu.mimuw.cloudatlas.model.PathName; -import pl.edu.mimuw.cloudatlas.model.Value; -import pl.edu.mimuw.cloudatlas.model.ValueContact; -import pl.edu.mimuw.cloudatlas.model.ValueSet; -import pl.edu.mimuw.cloudatlas.model.ZMI; - public class Agent { - private static EventBus eventBus; - private static GossipGirlStrategies.ZoneSelectionStrategy zoneSelectionStrategy; - private static GossipGirlStrategies gossipGirlStrategies; - private static Random random = new Random(); - - public static void runRegistry() { - try { - NewApiImplementation api = new NewApiImplementation(eventBus); - Api apiStub = - (Api) UnicastRemoteObject.exportObject(api, 0); - Registry registry = LocateRegistry.getRegistry(); - registry.rebind("Api", apiStub); - System.out.println("Agent: api bound"); - } catch (Exception e) { - System.err.println("Agent registry initialization exception:"); - e.printStackTrace(); - } - } - - public static HashMap initializeModules() throws UnknownHostException, SocketException, NullPointerException { - HashMap modules = new HashMap(); - modules.put(ModuleType.TIMER_SCHEDULER, new TimerScheduler(ModuleType.TIMER_SCHEDULER)); - modules.put(ModuleType.RMI, new Remik()); - Long freshnessPeriod = Long.getLong("freshness_period"); - modules.put(ModuleType.STATE, new Stanik(freshnessPeriod)); - modules.put(ModuleType.QUERY, new Qurnik()); - modules.put(ModuleType.GOSSIP, new GossipGirl()); - - Integer port = Integer.getInteger("UDUPServer.port"); - Integer timeout = Integer.getInteger("UDUPServer.timeout"); - Integer bufsize = Integer.getInteger("UDUPServer.bufsize"); - InetAddress serverAddr = InetAddress.getByName(System.getProperty("UDUPServer.hostname")); - UDUPServer server = new UDUPServer(serverAddr, port, bufsize); - modules.put(ModuleType.UDP, new UDUP(port, timeout, bufsize, server)); - return modules; - } - - public static HashMap initializeExecutors( - HashMap modules) { - HashMap executors = new HashMap(); - - for (Map.Entry moduleEntry : modules.entrySet()) { - Module module = moduleEntry.getValue(); - Executor executor = new Executor(module); - executors.put(moduleEntry.getKey(), executor); - } - - return executors; - } - - public static ArrayList initializeExecutorThreads(HashMap executors) { - ArrayList executorThreads = new ArrayList(); - - for (Map.Entry executorEntry : executors.entrySet()) { - Thread thread = new Thread(executorEntry.getValue()); - thread.setDaemon(true); - System.out.println("Initializing executor " + executorEntry.getKey()); - thread.start(); - executorThreads.add(thread); - } - - return executorThreads; - } - - public static void closeExecutors(ArrayList executorThreads) { - for (Thread executorThread : executorThreads) { - executorThread.interrupt(); - } - } - - public static void runModulesAsThreads() { - HashMap modules = null; - - try { - modules = initializeModules(); - } catch (UnknownHostException | SocketException e) { - System.out.println("Module initialization failed"); - e.printStackTrace(); - return; - } - - HashMap executors = initializeExecutors(modules); - ArrayList executorThreads = initializeExecutorThreads(executors); - eventBus = new EventBus(executors); - Thread UDUPServerThread = new Thread(((UDUP) modules.get(ModuleType.UDP)).getServer()); - Thread eventBusThread = new Thread(eventBus); - System.out.println("Initializing event bus"); - eventBusThread.start(); - UDUPServerThread.start(); - } - - private static void initZones() { - try { - ZMI root = Main.createTestHierarchy2(); - addZoneAndChildren(root, new PathName("")); - System.out.println("Initialized with test hierarchy"); - } catch (Exception e) { - System.out.println("ERROR: failed to create test hierarchy"); - } - } - private static void startQueries(long queriesPeriod) { - Supplier taskSupplier = () -> - new TimerScheduledTask() { - public void run() { - try { - sendMessage(new RunQueriesMessage("", 0)); - } catch (InterruptedException e) { - System.out.println("Interrupted while triggering queries"); - } - } - }; - - startRecursiveTask(taskSupplier, queriesPeriod); - } - - private static void startGossip(long gossipPeriod) { - Supplier taskSupplier = () -> - new TimerScheduledTask() { - public void run() { - try { - System.out.println("INFO: initiating gossip"); - PathName gossipLevel = gossipGirlStrategies.selectStrategy(zoneSelectionStrategy); - ValueContact contact = selectContactFromLevel(gossipLevel); - if (contact != null) { - System.out.println("INFO: found a contact " + contact.toString()); - InitiateGossipMessage message = new InitiateGossipMessage("", 0, new PathName("/uw/violet07"), contact); - sendMessage(message); - } else { - System.out.println("DEBUG: couldn't find contact for gossip"); - } - } catch (InterruptedException e) { - System.out.println("Interrupted while initiating gossip"); - } catch (Exception e) { - System.out.println("ERROR: something happened " + e.toString()); - } - } - }; - - startRecursiveTask(taskSupplier, gossipPeriod); - } - - private static ValueContact selectContactFromLevel(PathName path) throws Exception { - CompletableFuture responseFuture = new CompletableFuture(); - eventBus.addMessage(new RequestStateMessage("", 0, responseFuture)); - StateMessage response = (StateMessage) responseFuture.get(); - ZMI root = response.getZMI(); - List siblings = getSiblings(root, path); - filterEmptyContacts(siblings); - if (siblings.isEmpty()) { - return selectFallbackContact(); - } - ZMI zmi = selectZMI(siblings); - ValueSet contactsValue = (ValueSet) zmi.getAttributes().getOrNull("contacts"); - Set valueSet = contactsValue.getValue(); - return selectContactFromSet(valueSet); - } - - private static ValueContact selectFallbackContact() throws Exception { - return null; - } - - private static ZMI selectZMI(List zmis) throws Exception { - int i = random.nextInt(zmis.size()); - for (ZMI zmi : zmis) { - if (i == 0) { - return zmi; - } - i--; - } - System.out.println("ERROR: empty list passed to selectZMI"); - throw new Exception("empty list passed to selectZMI"); - } - - private static ValueContact selectContactFromSet(Set contacts) throws Exception { - int i = random.nextInt(contacts.size()); - for (Value contact : contacts) { - if (i == 0) { - return (ValueContact) contact; - } - i--; - } - System.out.println("ERROR: empty list passed to selectContactFromSet"); - throw new Exception("empty list passed to selectContactFromSet"); - } - - private static List getSiblings(ZMI root, PathName path) { - try { - List siblingsAndI = root.findDescendant(path).getFather().getSons(); - List siblings = new ArrayList(); - for (ZMI siblingOrI : siblingsAndI) { - if (!siblingOrI.getPathName().equals(path)) { - siblings.add(siblingOrI); - } - } - return siblings; - } catch (ZMI.NoSuchZoneException e) { - System.out.println("ERROR: didn't find path when looking for siblings"); - return new ArrayList(); - } - } - - private static void filterEmptyContacts(List zmis) { - Iterator iterator = zmis.iterator(); - while (iterator.hasNext()) { - ZMI zmi = iterator.next(); - ValueSet contacts = (ValueSet) zmi.getAttributes().getOrNull("contacts"); - if (contacts == null || contacts.isNull() || contacts.isEmpty()) { - iterator.remove(); - } - } - } - - private static void startRecursiveTask(Supplier taskSupplier, long period) { - TimerScheduledTask timerTask = new RecursiveScheduledTask(period, taskSupplier); + public static void main(String[] args) { + AgentConfig agentConfig = new AgentConfig(); - try { - eventBus.addMessage(new TimerSchedulerMessage("", 0, "", period, 0, timerTask)); - } catch (InterruptedException e) { - System.out.println("Interrupted while starting queries"); - } - } + agentConfig.runModulesAsThreads(); + agentConfig.runRegistry(); - private static void addZoneAndChildren(ZMI zmi, PathName pathName) { - try { - UpdateAttributesMessage message = new UpdateAttributesMessage("", 0, pathName.toString(), zmi.getAttributes()); - eventBus.addMessage(message); - for (ZMI son : zmi.getSons()) { - addZoneAndChildren(son, pathName.levelDown(son.getAttributes().getOrNull("name").toString())); - } - } catch (Exception e) { - System.out.println("ERROR: failed to add zone"); - } - } - - public static void main(String[] args) { - zoneSelectionStrategy = GossipGirlStrategies.ZoneSelectionStrategy.ROUND_ROBIN_SAME_FREQ; - gossipGirlStrategies = new GossipGirlStrategies(new PathName("/uw/violet07")); - runModulesAsThreads(); - runRegistry(); - initZones(); + HierarchyConfig hierarchyConfig = new HierarchyConfig(agentConfig.getEventBus()); + hierarchyConfig.initZones(); // TODO: make query period confiurable with config file and from tests Long queryPeriod = Long.getLong("query_period"); - startQueries(queryPeriod); + hierarchyConfig.startQueries(queryPeriod); Long gossipPeriod = Long.getLong("gossip_period"); - startGossip(gossipPeriod); + hierarchyConfig.startGossip(gossipPeriod); } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java new file mode 100644 index 0000000..b49525e --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java @@ -0,0 +1,113 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import pl.edu.mimuw.cloudatlas.agent.modules.*; +import pl.edu.mimuw.cloudatlas.agent.modules.Module; +import pl.edu.mimuw.cloudatlas.api.Api; + +import java.net.InetAddress; +import java.net.SocketException; +import java.net.UnknownHostException; +import java.rmi.registry.LocateRegistry; +import java.rmi.registry.Registry; +import java.rmi.server.UnicastRemoteObject; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; + +public class AgentConfig { + private EventBus eventBus; + + AgentConfig() {} + + public EventBus getEventBus() { + return eventBus; + } + + public void runRegistry() { + try { + NewApiImplementation api = new NewApiImplementation(eventBus); + Api apiStub = + (Api) UnicastRemoteObject.exportObject(api, 0); + Registry registry = LocateRegistry.getRegistry(); + registry.rebind("Api", apiStub); + System.out.println("Agent: api bound"); + } catch (Exception e) { + System.err.println("Agent registry initialization exception:"); + e.printStackTrace(); + } + } + + private HashMap initializeModules() throws UnknownHostException, SocketException, NullPointerException { + HashMap modules = new HashMap(); + modules.put(ModuleType.TIMER_SCHEDULER, new TimerScheduler(ModuleType.TIMER_SCHEDULER)); + modules.put(ModuleType.RMI, new Remik()); + Long freshnessPeriod = Long.getLong("freshness_period"); + modules.put(ModuleType.STATE, new Stanik(freshnessPeriod)); + modules.put(ModuleType.QUERY, new Qurnik()); + modules.put(ModuleType.GOSSIP, new GossipGirl()); + + Integer port = Integer.getInteger("UDUPServer.port"); + Integer timeout = Integer.getInteger("UDUPServer.timeout"); + Integer bufsize = Integer.getInteger("UDUPServer.bufsize"); + InetAddress serverAddr = InetAddress.getByName(System.getProperty("UDUPServer.hostname")); + UDUPServer server = new UDUPServer(serverAddr, port, bufsize); + modules.put(ModuleType.UDP, new UDUP(port, timeout, bufsize, server)); + return modules; + } + + private HashMap initializeExecutors( + HashMap modules) { + HashMap executors = new HashMap(); + + for (Map.Entry moduleEntry : modules.entrySet()) { + Module module = moduleEntry.getValue(); + Executor executor = new Executor(module); + executors.put(moduleEntry.getKey(), executor); + } + + return executors; + } + + private ArrayList initializeExecutorThreads(HashMap executors) { + ArrayList executorThreads = new ArrayList(); + + for (Map.Entry executorEntry : executors.entrySet()) { + Thread thread = new Thread(executorEntry.getValue()); + thread.setDaemon(true); + System.out.println("Initializing executor " + executorEntry.getKey()); + thread.start(); + executorThreads.add(thread); + } + + return executorThreads; + } + + public void closeExecutors(ArrayList executorThreads) { + for (Thread executorThread : executorThreads) { + executorThread.interrupt(); + } + } + + public void runModulesAsThreads() { + HashMap modules = null; + + try { + modules = initializeModules(); + } catch (UnknownHostException | SocketException e) { + System.out.println("Module initialization failed"); + e.printStackTrace(); + return; + } + + HashMap executors = initializeExecutors(modules); + ArrayList executorThreads = initializeExecutorThreads(executors); + eventBus = new EventBus(executors); + Thread UDUPServerThread = new Thread(((UDUP) modules.get(ModuleType.UDP)).getServer()); + Thread eventBusThread = new Thread(eventBus); + System.out.println("Initializing event bus"); + eventBusThread.start(); + UDUPServerThread.start(); + } + + +} diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java new file mode 100644 index 0000000..1ee9ac2 --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java @@ -0,0 +1,171 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import pl.edu.mimuw.cloudatlas.agent.messages.*; +import pl.edu.mimuw.cloudatlas.agent.modules.GossipGirl; +import pl.edu.mimuw.cloudatlas.agent.modules.GossipGirlStrategies; +import pl.edu.mimuw.cloudatlas.agent.modules.RecursiveScheduledTask; +import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduledTask; +import pl.edu.mimuw.cloudatlas.interpreter.Main; +import pl.edu.mimuw.cloudatlas.model.*; + +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +public class HierarchyConfig { + private GossipGirlStrategies gossipGirlStrategies; + private GossipGirlStrategies.ZoneSelectionStrategy zoneSelectionStrategy; + private Random random = new Random(); + private EventBus eventBus; + + HierarchyConfig(EventBus eventBus) { + zoneSelectionStrategy = GossipGirlStrategies.ZoneSelectionStrategy.ROUND_ROBIN_SAME_FREQ; + gossipGirlStrategies = new GossipGirlStrategies(new PathName("/uw/violet07")); + this.eventBus = eventBus; + } + + public void startGossip(long gossipPeriod) { + Supplier taskSupplier = () -> + new TimerScheduledTask() { + public void run() { + try { + System.out.println("INFO: initiating gossip"); + PathName gossipLevel = gossipGirlStrategies.selectStrategy(zoneSelectionStrategy); + ValueContact contact = selectContactFromLevel(gossipLevel); + if (contact != null) { + System.out.println("INFO: found a contact " + contact.toString()); + InitiateGossipMessage message = new InitiateGossipMessage("", 0, new PathName("/uw/violet07"), contact); + sendMessage(message); + } else { + System.out.println("DEBUG: couldn't find contact for gossip"); + } + } catch (InterruptedException e) { + System.out.println("Interrupted while initiating gossip"); + } catch (Exception e) { + System.out.println("ERROR: something happened " + e.toString()); + } + } + }; + + startRecursiveTask(taskSupplier, gossipPeriod); + } + + private ValueContact selectContactFromLevel(PathName path) throws Exception { + CompletableFuture responseFuture = new CompletableFuture(); + this.eventBus.addMessage(new RequestStateMessage("", 0, responseFuture)); + StateMessage response = (StateMessage) responseFuture.get(); + ZMI root = response.getZMI(); + List siblings = getSiblings(root, path); + filterEmptyContacts(siblings); + if (siblings.isEmpty()) { + return selectFallbackContact(); + } + ZMI zmi = selectZMI(siblings); + ValueSet contactsValue = (ValueSet) zmi.getAttributes().getOrNull("contacts"); + Set valueSet = contactsValue.getValue(); + return selectContactFromSet(valueSet); + } + + // TODO + private ValueContact selectFallbackContact() throws Exception { + return null; + } + + private ZMI selectZMI(List zmis) throws Exception { + int i = random.nextInt(zmis.size()); + for (ZMI zmi : zmis) { + if (i == 0) { + return zmi; + } + i--; + } + System.out.println("ERROR: empty list passed to selectZMI"); + throw new Exception("empty list passed to selectZMI"); + } + + private ValueContact selectContactFromSet(Set contacts) throws Exception { + int i = random.nextInt(contacts.size()); + for (Value contact : contacts) { + if (i == 0) { + return (ValueContact) contact; + } + i--; + } + System.out.println("ERROR: empty list passed to selectContactFromSet"); + throw new Exception("empty list passed to selectContactFromSet"); + } + + private List getSiblings(ZMI root, PathName path) { + try { + List siblingsAndI = root.findDescendant(path).getFather().getSons(); + List siblings = new ArrayList(); + for (ZMI siblingOrI : siblingsAndI) { + if (!siblingOrI.getPathName().equals(path)) { + siblings.add(siblingOrI); + } + } + return siblings; + } catch (ZMI.NoSuchZoneException e) { + System.out.println("ERROR: didn't find path when looking for siblings"); + return new ArrayList(); + } + } + + private void filterEmptyContacts(List zmis) { + Iterator iterator = zmis.iterator(); + while (iterator.hasNext()) { + ZMI zmi = iterator.next(); + ValueSet contacts = (ValueSet) zmi.getAttributes().getOrNull("contacts"); + if (contacts == null || contacts.isNull() || contacts.isEmpty()) { + iterator.remove(); + } + } + } + + public void startRecursiveTask(Supplier taskSupplier, long period) { + TimerScheduledTask timerTask = new RecursiveScheduledTask(period, taskSupplier); + + try { + this.eventBus.addMessage(new TimerSchedulerMessage("", 0, "", period, 0, timerTask)); + } catch (InterruptedException e) { + System.out.println("Interrupted while starting queries"); + } + } + + private void addZoneAndChildren(ZMI zmi, PathName pathName) { + try { + UpdateAttributesMessage message = new UpdateAttributesMessage("", 0, pathName.toString(), zmi.getAttributes()); + this.eventBus.addMessage(message); + for (ZMI son : zmi.getSons()) { + addZoneAndChildren(son, pathName.levelDown(son.getAttributes().getOrNull("name").toString())); + } + } catch (Exception e) { + System.out.println("ERROR: failed to add zone"); + } + } + + public void initZones() { + try { + ZMI root = Main.createTestHierarchy2(); + addZoneAndChildren(root, new PathName("")); + System.out.println("Initialized with test hierarchy"); + } catch (Exception e) { + System.out.println("ERROR: failed to create test hierarchy"); + } + } + + public void startQueries(long queriesPeriod) { + Supplier taskSupplier = () -> + new TimerScheduledTask() { + public void run() { + try { + sendMessage(new RunQueriesMessage("", 0)); + } catch (InterruptedException e) { + System.out.println("Interrupted while triggering queries"); + } + } + }; + + startRecursiveTask(taskSupplier, queriesPeriod); + } +} -- cgit v1.2.3 From 22ec12d762db07b6ecc65462b1e7cb38c8ca1578 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Magdalena=20Grodzi=C5=84ska?= Date: Sat, 11 Jan 2020 20:06:14 +0100 Subject: Refactor agent configs nicely --- .../java/pl/edu/mimuw/cloudatlas/agent/Agent.java | 44 ++++++++++++++-- .../pl/edu/mimuw/cloudatlas/agent/AgentConfig.java | 33 ++++++------ .../pl/edu/mimuw/cloudatlas/agent/AgentUtils.java | 19 +++++++ .../pl/edu/mimuw/cloudatlas/agent/EventBus.java | 5 ++ .../mimuw/cloudatlas/agent/HierarchyConfig.java | 61 ++++++++-------------- 5 files changed, 101 insertions(+), 61 deletions(-) create mode 100644 src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentUtils.java (limited to 'src/main/java/pl/edu') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java index 690198b..a7d2d55 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java @@ -1,19 +1,55 @@ package pl.edu.mimuw.cloudatlas.agent; +import pl.edu.mimuw.cloudatlas.agent.messages.UpdateAttributesMessage; +import pl.edu.mimuw.cloudatlas.interpreter.Main; +import pl.edu.mimuw.cloudatlas.model.PathName; +import pl.edu.mimuw.cloudatlas.model.ZMI; + public class Agent { + private static void addZoneAndChildren(ZMI zmi, PathName pathName, EventBus eventBus) { + try { + UpdateAttributesMessage message = new UpdateAttributesMessage("", 0, pathName.toString(), zmi.getAttributes()); + eventBus.addMessage(message); + for (ZMI son : zmi.getSons()) { + addZoneAndChildren(son, pathName.levelDown(son.getAttributes().getOrNull("name").toString()), eventBus); + } + } catch (Exception e) { + System.out.println("ERROR: failed to add zone"); + } + } + + public static void initZones(EventBus eventBus) { + try { + ZMI root = Main.createTestHierarchy2(); + addZoneAndChildren(root, new PathName(""), eventBus); + System.out.println("Initialized with test hierarchy"); + } catch (Exception e) { + System.out.println("ERROR: failed to create test hierarchy"); + } + } + public static void main(String[] args) { AgentConfig agentConfig = new AgentConfig(); agentConfig.runModulesAsThreads(); - agentConfig.runRegistry(); - HierarchyConfig hierarchyConfig = new HierarchyConfig(agentConfig.getEventBus()); - hierarchyConfig.initZones(); + EventBus eventBus = new EventBus(agentConfig.getExecutors()); + agentConfig.runRegistry(eventBus); + agentConfig.startNonModuleThreads(eventBus); + + initZones(eventBus); + // TODO: make query period confiurable with config file and from tests + + // TODO config setup + String zonePath = System.getProperty("zone_path"); + String selectionStrategy = System.getProperty("Gossip.zone_strategy"); Long queryPeriod = Long.getLong("query_period"); - hierarchyConfig.startQueries(queryPeriod); Long gossipPeriod = Long.getLong("gossip_period"); + + HierarchyConfig hierarchyConfig = new HierarchyConfig(eventBus, zonePath, selectionStrategy); + hierarchyConfig.startQueries(queryPeriod); hierarchyConfig.startGossip(gossipPeriod); } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java index b49525e..16bd99a 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java @@ -15,15 +15,14 @@ import java.util.HashMap; import java.util.Map; public class AgentConfig { - private EventBus eventBus; + private HashMap executors; + HashMap modules; - AgentConfig() {} - - public EventBus getEventBus() { - return eventBus; + public HashMap getExecutors() { + return executors; } - public void runRegistry() { + public void runRegistry(EventBus eventBus) { try { NewApiImplementation api = new NewApiImplementation(eventBus); Api apiStub = @@ -38,18 +37,20 @@ public class AgentConfig { } private HashMap initializeModules() throws UnknownHostException, SocketException, NullPointerException { + // TODO config setup + Long freshnessPeriod = Long.getLong("freshness_period"); + Integer port = Integer.getInteger("UDUPServer.port"); + Integer timeout = Integer.getInteger("UDUPServer.timeout"); + Integer bufsize = Integer.getInteger("UDUPServer.bufsize"); + InetAddress serverAddr = InetAddress.getByName(System.getProperty("UDUPServer.hostname")); + HashMap modules = new HashMap(); modules.put(ModuleType.TIMER_SCHEDULER, new TimerScheduler(ModuleType.TIMER_SCHEDULER)); modules.put(ModuleType.RMI, new Remik()); - Long freshnessPeriod = Long.getLong("freshness_period"); modules.put(ModuleType.STATE, new Stanik(freshnessPeriod)); modules.put(ModuleType.QUERY, new Qurnik()); modules.put(ModuleType.GOSSIP, new GossipGirl()); - Integer port = Integer.getInteger("UDUPServer.port"); - Integer timeout = Integer.getInteger("UDUPServer.timeout"); - Integer bufsize = Integer.getInteger("UDUPServer.bufsize"); - InetAddress serverAddr = InetAddress.getByName(System.getProperty("UDUPServer.hostname")); UDUPServer server = new UDUPServer(serverAddr, port, bufsize); modules.put(ModuleType.UDP, new UDUP(port, timeout, bufsize, server)); return modules; @@ -89,8 +90,6 @@ public class AgentConfig { } public void runModulesAsThreads() { - HashMap modules = null; - try { modules = initializeModules(); } catch (UnknownHostException | SocketException e) { @@ -99,15 +98,15 @@ public class AgentConfig { return; } - HashMap executors = initializeExecutors(modules); + executors = initializeExecutors(modules); ArrayList executorThreads = initializeExecutorThreads(executors); - eventBus = new EventBus(executors); + } + + void startNonModuleThreads(EventBus eventBus) { Thread UDUPServerThread = new Thread(((UDUP) modules.get(ModuleType.UDP)).getServer()); Thread eventBusThread = new Thread(eventBus); System.out.println("Initializing event bus"); eventBusThread.start(); UDUPServerThread.start(); } - - } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentUtils.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentUtils.java new file mode 100644 index 0000000..62d4bcd --- /dev/null +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentUtils.java @@ -0,0 +1,19 @@ +package pl.edu.mimuw.cloudatlas.agent; + +import pl.edu.mimuw.cloudatlas.agent.messages.TimerSchedulerMessage; +import pl.edu.mimuw.cloudatlas.agent.modules.RecursiveScheduledTask; +import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduledTask; + +import java.util.function.Supplier; + +public class AgentUtils { + + public static void startRecursiveTask(Supplier taskSupplier, long period, EventBus eventBus) { + TimerScheduledTask timerTask = new RecursiveScheduledTask(period, taskSupplier); + try { + eventBus.addMessage(new TimerSchedulerMessage("", 0, "", period, 0, timerTask)); + } catch (InterruptedException e) { + System.out.println("Interrupted while starting queries"); + } + } +} diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java index a6d3b2d..a9b25ed 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java @@ -22,6 +22,7 @@ public class EventBus implements Runnable { // Allows for testing with a mock EventBus protected EventBus() { + this.events = new LinkedBlockingQueue(); } EventBus(HashMap executors) { @@ -30,6 +31,10 @@ public class EventBus implements Runnable { this.events = new LinkedBlockingQueue(); } + public void setEvents(LinkedBlockingQueue events) { + this.events = events; + } + public void run() { System.out.println("Event bus running"); while (!Thread.currentThread().interrupted()) { diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java index 1ee9ac2..04b9b5b 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java @@ -1,13 +1,11 @@ package pl.edu.mimuw.cloudatlas.agent; import pl.edu.mimuw.cloudatlas.agent.messages.*; -import pl.edu.mimuw.cloudatlas.agent.modules.GossipGirl; import pl.edu.mimuw.cloudatlas.agent.modules.GossipGirlStrategies; -import pl.edu.mimuw.cloudatlas.agent.modules.RecursiveScheduledTask; import pl.edu.mimuw.cloudatlas.agent.modules.TimerScheduledTask; -import pl.edu.mimuw.cloudatlas.interpreter.Main; import pl.edu.mimuw.cloudatlas.model.*; +import java.lang.reflect.UndeclaredThrowableException; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.function.Supplier; @@ -18,12 +16,27 @@ public class HierarchyConfig { private Random random = new Random(); private EventBus eventBus; - HierarchyConfig(EventBus eventBus) { - zoneSelectionStrategy = GossipGirlStrategies.ZoneSelectionStrategy.ROUND_ROBIN_SAME_FREQ; - gossipGirlStrategies = new GossipGirlStrategies(new PathName("/uw/violet07")); + HierarchyConfig(EventBus eventBus, String zonePath, String zoneStrategy) { + zoneSelectionStrategy = parseStrategy(zoneStrategy); + gossipGirlStrategies = new GossipGirlStrategies(new PathName(zonePath)); this.eventBus = eventBus; } + private GossipGirlStrategies.ZoneSelectionStrategy parseStrategy(String selectionStrategy) { + switch (selectionStrategy) { + case "RoundRobinExp": + return GossipGirlStrategies.ZoneSelectionStrategy.ROUND_ROBIN_EXP_FREQ; + case "RoundRobinUniform": + return GossipGirlStrategies.ZoneSelectionStrategy.ROUND_ROBIN_SAME_FREQ; + case "RandomExp": + return GossipGirlStrategies.ZoneSelectionStrategy.RANDOM_DECR_EXP; + case "RandomUniform": + return GossipGirlStrategies.ZoneSelectionStrategy.RANDOM_UNFIORM; + default: + throw new UnsupportedOperationException("Selection strategy doesnt exist"); + } + } + public void startGossip(long gossipPeriod) { Supplier taskSupplier = () -> new TimerScheduledTask() { @@ -47,7 +60,7 @@ public class HierarchyConfig { } }; - startRecursiveTask(taskSupplier, gossipPeriod); + AgentUtils.startRecursiveTask(taskSupplier, gossipPeriod, eventBus); } private ValueContact selectContactFromLevel(PathName path) throws Exception { @@ -122,38 +135,6 @@ public class HierarchyConfig { } } - public void startRecursiveTask(Supplier taskSupplier, long period) { - TimerScheduledTask timerTask = new RecursiveScheduledTask(period, taskSupplier); - - try { - this.eventBus.addMessage(new TimerSchedulerMessage("", 0, "", period, 0, timerTask)); - } catch (InterruptedException e) { - System.out.println("Interrupted while starting queries"); - } - } - - private void addZoneAndChildren(ZMI zmi, PathName pathName) { - try { - UpdateAttributesMessage message = new UpdateAttributesMessage("", 0, pathName.toString(), zmi.getAttributes()); - this.eventBus.addMessage(message); - for (ZMI son : zmi.getSons()) { - addZoneAndChildren(son, pathName.levelDown(son.getAttributes().getOrNull("name").toString())); - } - } catch (Exception e) { - System.out.println("ERROR: failed to add zone"); - } - } - - public void initZones() { - try { - ZMI root = Main.createTestHierarchy2(); - addZoneAndChildren(root, new PathName("")); - System.out.println("Initialized with test hierarchy"); - } catch (Exception e) { - System.out.println("ERROR: failed to create test hierarchy"); - } - } - public void startQueries(long queriesPeriod) { Supplier taskSupplier = () -> new TimerScheduledTask() { @@ -166,6 +147,6 @@ public class HierarchyConfig { } }; - startRecursiveTask(taskSupplier, queriesPeriod); + AgentUtils.startRecursiveTask(taskSupplier, queriesPeriod, eventBus); } } -- cgit v1.2.3 From 3831d80df5dcd5d75d95308c8906dd9e3c054334 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Magdalena=20Grodzi=C5=84ska?= Date: Sat, 11 Jan 2020 20:12:03 +0100 Subject: Fix tests --- src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src/main/java/pl/edu') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java index 16bd99a..38d764a 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/AgentConfig.java @@ -56,7 +56,7 @@ public class AgentConfig { return modules; } - private HashMap initializeExecutors( + public static HashMap initializeExecutors( HashMap modules) { HashMap executors = new HashMap(); @@ -69,7 +69,7 @@ public class AgentConfig { return executors; } - private ArrayList initializeExecutorThreads(HashMap executors) { + public static ArrayList initializeExecutorThreads(HashMap executors) { ArrayList executorThreads = new ArrayList(); for (Map.Entry executorEntry : executors.entrySet()) { -- cgit v1.2.3 From c631dee52c21df09816ece63d6e6816a408da2f5 Mon Sep 17 00:00:00 2001 From: Marcin Chrzanowski Date: Sat, 11 Jan 2020 21:02:34 +0100 Subject: Fix things --- src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java | 2 +- src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java | 1 + .../pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java | 14 +++++++++----- .../edu/mimuw/cloudatlas/agent/messages/StateMessage.java | 10 +++++++++- .../java/pl/edu/mimuw/cloudatlas/agent/modules/Stanik.java | 6 ++++-- 5 files changed, 24 insertions(+), 9 deletions(-) (limited to 'src/main/java/pl/edu') diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java index a7d2d55..0cbda2d 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java @@ -38,7 +38,7 @@ public class Agent { agentConfig.runRegistry(eventBus); agentConfig.startNonModuleThreads(eventBus); - initZones(eventBus); + // initZones(eventBus); // TODO: make query period confiurable with config file and from tests diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java index a9b25ed..0e972f4 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/EventBus.java @@ -54,6 +54,7 @@ public class EventBus implements Runnable { } public void addMessage(AgentMessage msg) throws InterruptedException { + System.out.println("INFO: message added for " + msg.getDestinationModule().toString()); this.events.put(msg); } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java index 04b9b5b..dc5241d 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/HierarchyConfig.java @@ -71,7 +71,7 @@ public class HierarchyConfig { List siblings = getSiblings(root, path); filterEmptyContacts(siblings); if (siblings.isEmpty()) { - return selectFallbackContact(); + return selectFallbackContact(response.getContacts()); } ZMI zmi = selectZMI(siblings); ValueSet contactsValue = (ValueSet) zmi.getAttributes().getOrNull("contacts"); @@ -80,8 +80,12 @@ public class HierarchyConfig { } // TODO - private ValueContact selectFallbackContact() throws Exception { - return null; + private ValueContact selectFallbackContact(Set contacts) throws Exception { + if (contacts.isEmpty()) { + return null; + } else { + return selectContactFromSet(contacts); + } } private ZMI selectZMI(List zmis) throws Exception { @@ -96,9 +100,9 @@ public class HierarchyConfig { throw new Exception("empty list passed to selectZMI"); } - private ValueContact selectContactFromSet(Set contacts) throws Exception { + private ValueContact selectContactFromSet(Set contacts) throws Exception { int i = random.nextInt(contacts.size()); - for (Value contact : contacts) { + for (T contact : contacts) { if (i == 0) { return (ValueContact) contact; } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/StateMessage.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/StateMessage.java index 759723b..806d41f 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/StateMessage.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/StateMessage.java @@ -2,9 +2,11 @@ package pl.edu.mimuw.cloudatlas.agent.messages; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; import pl.edu.mimuw.cloudatlas.agent.modules.ModuleType; import pl.edu.mimuw.cloudatlas.model.Attribute; +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; @@ -12,11 +14,13 @@ import pl.edu.mimuw.cloudatlas.model.ZMI; public class StateMessage extends ResponseMessage { private ZMI zmi; private Map> queries; + private Set contacts; - public StateMessage(String messageId, ModuleType destinationModule, long timestamp, long requestId, ZMI zmi, Map> queries) { + public StateMessage(String messageId, ModuleType destinationModule, long timestamp, long requestId, ZMI zmi, Map> queries, Set contacts) { super(messageId, destinationModule, timestamp, Type.STATE, requestId); this.zmi = zmi; this.queries = queries; + this.contacts = contacts; } public StateMessage() {} @@ -28,4 +32,8 @@ public class StateMessage extends ResponseMessage { public Map> getQueries() { return queries; } + + public Set getContacts() { + return contacts; + } } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/Stanik.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/Stanik.java index c713827..999c193 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/Stanik.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/Stanik.java @@ -53,8 +53,9 @@ public class Stanik extends Module { break; case UPDATE_CONTACTS: handleUpdateContacts((UpdateContactsMessage) message); + break; default: - throw new InvalidMessageType("This type of message cannot be handled by Stanik"); + throw new InvalidMessageType("This type of message cannot be handled by Stanik" + message.getType().toString()); } } @@ -66,7 +67,8 @@ public class Stanik extends Module { 0, message.getRequestId(), hierarchy.clone(), - (HashMap>) queries.clone() + (HashMap>) queries.clone(), + contacts ); sendMessage(response); } -- cgit v1.2.3