1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
package pl.edu.mimuw.cloudatlas.agent;
import java.net.Inet4Address;
import java.net.InetAddress;
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;
import java.util.function.Supplier;
import pl.edu.mimuw.cloudatlas.agent.NewApiImplementation;
import pl.edu.mimuw.cloudatlas.agent.messages.RunQueriesMessage;
import pl.edu.mimuw.cloudatlas.agent.messages.TimerSchedulerMessage;
import pl.edu.mimuw.cloudatlas.agent.messages.UpdateAttributesMessage;
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.ZMI;
public class Agent {
private static EventBus eventBus;
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<ModuleType, Module> initializeModules() {
HashMap<ModuleType, Module> modules = new HashMap<ModuleType, Module>();
modules.put(ModuleType.TIMER_SCHEDULER, new TimerScheduler(ModuleType.TIMER_SCHEDULER));
modules.put(ModuleType.RMI, new Remik());
Long freshnessPeriod = new Long(System.getProperty("freshness_period"));
modules.put(ModuleType.STATE, new Stanik(freshnessPeriod));
modules.put(ModuleType.QUERY, new Qurnik());
try {
modules.put(ModuleType.UDP, new UDUP(InetAddress.getByName("127.0.0.1"), 5988, 5000, 20000));
} catch (UnknownHostException e) {
e.printStackTrace();
}
// TODO add modules as we implement them
return modules;
}
public static HashMap<ModuleType, Executor> initializeExecutors(
HashMap<ModuleType, Module> modules) {
HashMap<ModuleType, Executor> executors = new HashMap<ModuleType, Executor>();
for (Map.Entry<ModuleType, Module> moduleEntry : modules.entrySet()) {
Module module = moduleEntry.getValue();
Executor executor = new Executor(module);
executors.put(moduleEntry.getKey(), executor);
}
return executors;
}
public static ArrayList<Thread> initializeExecutorThreads(HashMap<ModuleType, Executor> executors) {
ArrayList<Thread> executorThreads = new ArrayList<Thread>();
for (Map.Entry<ModuleType, Executor> 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<Thread> executorThreads) {
for (Thread executorThread : executorThreads) {
executorThread.interrupt();
}
}
public static void runModulesAsThreads() {
HashMap<ModuleType, Module> modules = initializeModules();
HashMap<ModuleType, Executor> executors = initializeExecutors(modules);
ArrayList<Thread> executorThreads = initializeExecutorThreads(executors);
eventBus = new EventBus(executors);
Thread eventBusThread = new Thread(eventBus);
System.out.println("Initializing event bus");
eventBusThread.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<TimerScheduledTask> taskSupplier = () ->
new TimerScheduledTask() {
public void run() {
try {
sendMessage(new RunQueriesMessage("", 0));
} catch (InterruptedException e) {
System.out.println("Interrupted while triggering queries");
}
}
};
TimerScheduledTask timerTask = new RecursiveScheduledTask(queriesPeriod, taskSupplier);
try {
eventBus.addMessage(new TimerSchedulerMessage("", 0, "", queriesPeriod, 0, timerTask));
} catch (InterruptedException e) {
System.out.println("Interrupted while starting queries");
}
}
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) {
runModulesAsThreads();
runRegistry();
initZones();
// TODO: make query period confiurable with config file and from tests
startQueries(100l);
}
}
|