Skip to content

Commit e7abe5c

Browse files
lesquoyblesquoyb
committed
[GS] Adds threaded command management + custom keepalive
GS seems to have its own automatic keepalive, but there's no way of tuning it properly, hence the custom one that runs in parallel. To deactivate, set the ping_interval parameter to a negative number
1 parent 3a54d19 commit e7abe5c

8 files changed

Lines changed: 109 additions & 25 deletions

File tree

‎msi.gama.headless/src/msi/gama/headless/job/ManualExperimentJob.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,7 @@ public void initParam(final GamaJsonList p) {
169169
}
170170

171171
/**
172-
* Inits the end contion.
172+
* Inits the end condition.
173173
*
174174
* @param cond
175175
* the cond

‎msi.gama.headless/src/msi/gama/headless/listener/CommandExecutor.java‎

Lines changed: 39 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,36 @@
11
package msi.gama.headless.listener;
22

3+
import java.util.AbstractMap;
34
import java.util.Collections;
45
import java.util.HashMap;
6+
import java.util.LinkedList;
57
import java.util.Map;
8+
import java.util.Map.Entry;
9+
import java.util.Queue;
610

711
import org.java_websocket.WebSocket;
812
import org.java_websocket.enums.ReadyState;
913

14+
import msi.gama.runtime.concurrent.GamaExecutorService;
1015
import msi.gama.util.IMap;
1116
import msi.gama.util.file.json.Jsoner;
1217

1318
public class CommandExecutor {
1419

15-
private final Map<String, ISocketCommand> COMMANDS;
20+
protected final Map<String, ISocketCommand> COMMANDS;
1621

22+
protected volatile Queue<Entry<WebSocket, IMap<String, Object>>> commandQueue;
23+
24+
protected final Thread commandExecutionThread = new Thread(() -> {
25+
while (true) {
26+
while(!commandQueue.isEmpty()) {
27+
var cmd = commandQueue.poll();
28+
process(cmd.getKey(), cmd.getValue());
29+
}
30+
}
31+
});
32+
33+
1734
public CommandExecutor() {
1835
final Map<String, ISocketCommand> cmds = new HashMap<>();
1936
cmds.put("load", new LoadCommand());
@@ -31,21 +48,36 @@ public CommandExecutor() {
3148
cmds.put("fetch", new FetchCommand());
3249

3350
COMMANDS = Collections.unmodifiableMap(cmds);
51+
52+
commandQueue = new LinkedList<Map.Entry<WebSocket,IMap<String,Object>>>();
53+
commandExecutionThread.setUncaughtExceptionHandler(GamaExecutorService.EXCEPTION_HANDLER);
54+
commandExecutionThread.start();
3455
}
3556

36-
public void process(final WebSocket socket, final IMap<String, Object> map) {
57+
public void pushCommand(final WebSocket socket, final IMap<String, Object> map) {
58+
commandQueue.add(new AbstractMap.SimpleEntry<WebSocket, IMap<String, Object>>(socket, map));
59+
}
60+
61+
protected void process(final WebSocket socket, final IMap<String, Object> map) {
3762
final String cmd_type = map.get("type").toString();
3863
ISocketCommand command = COMMANDS.get(cmd_type);
3964

4065
if (command == null) {
4166
throw new IllegalArgumentException("Invalid command type: " + cmd_type);
4267
}
4368

44-
var res = command.execute(socket, map);
45-
if(res!=null) {
46-
if(socket.getReadyState().equals(ReadyState.OPEN))
47-
socket.send(Jsoner.serialize(res));
48-
}
69+
// Executes the command in a separate thread so the executor can
70+
// continue with the next one without waiting for it to finish
71+
new Thread(() -> {
72+
var res = command.execute(socket, map);
73+
if(res!=null) {
74+
if(socket.getReadyState().equals(ReadyState.OPEN)) {
75+
socket.send(Jsoner.serialize(res));
76+
}
77+
}
78+
}).start();
4979
}
5080

81+
82+
5183
}

‎msi.gama.headless/src/msi/gama/headless/listener/FetchCommand.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,6 @@ public class FetchCommand implements ISocketCommand {
2929
@Override
3030
public CommandResponse execute(final WebSocket socket, final IMap<String, Object> map) {
3131
final String exp_id = map.get("exp_id") != null ? map.get("exp_id").toString() : "";
32-
final String socket_id =
33-
map.get("socket_id") != null ? map.get("socket_id").toString() : "" + socket.hashCode();
3432

3533
final Object filepath = map.get("file");
3634
final Object access = map.get("access");

‎msi.gama.headless/src/msi/gama/headless/listener/GamaListener.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ public ConcurrentHashMap<String, ConcurrentHashMap<String, ManualExperimentJob>>
3232

3333
private static PrintStream errorStream;
3434

35-
public GamaListener(final int p, final Application a, final boolean secure, final String jksPath, final String spwd, final String kpwd) {
35+
public GamaListener(final int p, final Application a, final boolean secure, final String jksPath, final String spwd, final String kpwd, final int ping_interval) {
3636
File currentJavaJarFile = new File(
3737
GamaListener.class.getProtectionDomain().getCodeSource().getLocation().getPath());
3838
String currentJavaJarFilePath = currentJavaJarFile.getAbsolutePath();
@@ -41,16 +41,16 @@ public GamaListener(final int p, final Application a, final boolean secure, fina
4141

4242
Globals.IMAGES_PATH = Globals.TEMP_PATH + "\\snapshot";
4343
GAMA.setHeadLessMode(true, new GamaServerGUIHandler()); //todo: done here and in headless simulation loader, should be refactored
44-
createSocketServer(p, a, secure, jksPath, spwd, kpwd);
44+
createSocketServer(p, a, secure, jksPath, spwd, kpwd, ping_interval);
4545
}
4646

4747
/**
4848
* Creates the socket server.
4949
*
5050
* @throws UnknownHostException the unknown host exception
5151
*/
52-
public void createSocketServer(final int port, final Application a, final boolean ssl, final String jksPath, final String spwd, final String kpwd) {
53-
instance = new GamaWebSocketServer(port, a, this, ssl,jksPath,spwd,kpwd);
52+
public void createSocketServer(final int port, final Application a, final boolean ssl, final String jksPath, final String spwd, final String kpwd, final int ping_interval) {
53+
instance = new GamaWebSocketServer(port, a, this, ssl,jksPath,spwd,kpwd, ping_interval);
5454
instance.start();
5555
System.out.println("Gama Listener started on port: " + instance.getPort());
5656

‎msi.gama.headless/src/msi/gama/headless/listener/GamaWebSocketServer.java‎

Lines changed: 48 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,14 +21,19 @@
2121
import java.net.InetSocketAddress;
2222
import java.nio.ByteBuffer;
2323
import java.security.KeyStore;
24+
import java.util.HashMap;
2425
import java.util.List;
26+
import java.util.Map;
27+
import java.util.Timer;
28+
import java.util.TimerTask;
2529

2630
import javax.net.ssl.KeyManagerFactory;
2731
import javax.net.ssl.SSLContext;
2832
import javax.net.ssl.SSLParameters;
2933
import javax.net.ssl.TrustManagerFactory;
3034

3135
import org.java_websocket.WebSocket;
36+
import org.java_websocket.framing.Framedata;
3237
import org.java_websocket.handshake.ClientHandshake;
3338
import org.java_websocket.server.SSLParametersWebSocketServerFactory;
3439
import org.java_websocket.server.WebSocketServer;
@@ -44,7 +49,6 @@
4449
import msi.gama.headless.runtime.Application;
4550
import msi.gama.headless.script.ExperimentationPlanFactory;
4651
import msi.gama.util.GamaMapFactory;
47-
import msi.gama.util.IList;
4852
import msi.gama.util.IMap;
4953
import msi.gama.util.file.json.Jsoner;
5054
import ummisco.gama.dev.utils.DEBUG;
@@ -82,6 +86,11 @@ public void set_listener(final GamaListener _listener) {
8286
/** The cmd helper. */
8387
CommandExecutor cmdHelper;
8488

89+
// variables for the keepalive pings
90+
public final boolean canPing; // false if pingInterval is negative
91+
public final int pingInterval; // the time interval between two ping requests in ms
92+
protected Map<WebSocket, Timer> pingTimers; // map of all connected clients and their associated timers running ping requests
93+
8594
/**
8695
* Instantiates a new gama web socket server.
8796
*
@@ -94,8 +103,14 @@ public void set_listener(final GamaListener _listener) {
94103
* @param ssl
95104
* the ssl
96105
*/
97-
public GamaWebSocketServer(final int port, final Application a, final GamaListener l, final boolean ssl, final String jksPath, final String spwd, final String kpwd) {
106+
public GamaWebSocketServer(final int port, final Application a, final GamaListener l, final boolean ssl, final String jksPath, final String spwd, final String kpwd, final int ping_interval) {
98107
super(new InetSocketAddress(port));
108+
109+
canPing = ping_interval >= 0;
110+
pingInterval = ping_interval;
111+
pingTimers = new HashMap<WebSocket, Timer>();
112+
113+
99114
if (a.verbose) { DEBUG.ON(); }
100115
cmdHelper = new CommandExecutor();
101116
if (ssl) {
@@ -148,7 +163,17 @@ public void onOpen(final WebSocket conn, final ClientHandshake handshake) {
148163
DEBUG.OUT(conn.getRemoteSocketAddress().getAddress().getHostAddress() + " entered the room!");
149164
conn.send(Jsoner
150165
.serialize(new GamaServerMessage(GamaServerMessageType.ConnectionSuccessful, "" + conn.hashCode())));
151-
166+
167+
if (canPing) {
168+
var timer = new Timer();
169+
timer.scheduleAtFixedRate(new TimerTask() {
170+
@Override
171+
public void run() {
172+
conn.sendPing();
173+
}
174+
}, 0, pingInterval);
175+
pingTimers.put(conn, timer);
176+
}
152177
// String path = URI.create(handshake.getResourceDescriptor()).getPath();
153178
}
154179

@@ -159,8 +184,26 @@ public void onOpen(final WebSocket conn, final ClientHandshake handshake) {
159184
*/
160185
public Application getDefaultApp() { return app; }
161186

187+
@Override
188+
public void onWebsocketPing(WebSocket conn, Framedata f) {
189+
// TODO Auto-generated method stub
190+
super.onWebsocketPing(conn, f);
191+
}
192+
193+
@Override
194+
public void onWebsocketPong(WebSocket conn, Framedata f) {
195+
// TODO Auto-generated method stub
196+
super.onWebsocketPong(conn, f);
197+
}
198+
162199
@Override
163200
public void onClose(final WebSocket conn, final int code, final String reason, final boolean remote) {
201+
202+
var timer = pingTimers.remove(conn);
203+
if (timer != null) {
204+
timer.cancel();
205+
}
206+
164207
if (_listener.getLaunched_experiments().get("" + conn.hashCode()) != null) {
165208
for (ManualExperimentJob e : _listener.getLaunched_experiments().get("" + conn.hashCode()).values()) {
166209
e.controller.directPause();
@@ -217,13 +260,13 @@ public void onMessage(final WebSocket socket, final String message) {
217260
final String socket_id = map.get("socket_id") != null ? map.get("socket_id").toString() : ("" + socket.hashCode());
218261
if(get_listener().getExperiment(socket_id, exp_id)!=null && !get_listener().getExperiment(socket_id, exp_id).controller.isPaused() ) {
219262
get_listener().getExperiment(socket_id, exp_id).controller.getScope().getSimulation().postOneShotAction(scope1 -> {
220-
cmdHelper.process(socket, map);
263+
cmdHelper.pushCommand(socket, map);
221264
// System.out.println(map.get("type"));
222265
return null;
223266
});
224267
}else {
225268

226-
cmdHelper.process(socket, map);
269+
cmdHelper.pushCommand(socket, map);
227270
}
228271

229272
} catch (Exception e1) {

‎msi.gama.headless/src/msi/gama/headless/listener/ServerExperimentController.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,9 @@
1515
import java.util.concurrent.Semaphore;
1616

1717
import org.java_websocket.WebSocket;
18-
import org.locationtech.jts.util.Debug;
1918

2019
import msi.gama.common.interfaces.IGui;
2120
import msi.gama.headless.core.GamaHeadlessException;
22-
import msi.gama.headless.core.GamaServerMessage;
2321
import msi.gama.headless.core.GamaServerMessageType;
2422
import msi.gama.headless.job.ManualExperimentJob;
2523
import msi.gama.kernel.experiment.ExperimentAgent;

‎msi.gama.headless/src/msi/gama/headless/listener/StepCommand.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public CommandResponse execute(final WebSocket socket, IMap<String, Object> map)
2929
for (int i = 0 ; i < nb_step ; i++) {
3030
try {
3131
if (sync) {
32-
gama_exp.controller._job.doStep();
32+
gama_exp.doStep();
3333
} else {
3434
gama_exp.controller.userStep();
3535
}

‎msi.gama.headless/src/msi/gama/headless/runtime/Application.java‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,8 @@ public class Application implements IApplication {
8383

8484
/** The Constant THREAD_PARAMETER. */
8585
final public static String THREAD_PARAMETER = "-hpc";
86+
87+
final public static String PING_INTERVAL = "-ping_interval";
8688

8789
/** The Constant SOCKET_PARAMETER. */
8890
final public static String SOCKET_PARAMETER = "-socket";
@@ -133,6 +135,9 @@ public class Application implements IApplication {
133135
/** The socket. */
134136
public int socket = -1;
135137

138+
// the interval between each ping sent by the server, -1 to deactivate this behaviour
139+
public int ping_interval = 10000;
140+
136141
/** The console mode. */
137142
public boolean consoleMode = false;
138143

@@ -168,9 +173,13 @@ private static void showHelp() {
168173
+ " -- start the console to write xml parameter file" + "\n\t\t"
169174
+ VERBOSE_PARAMETER + " -- verbose mode" + "\n\t\t" + THREAD_PARAMETER
170175
+ " [core] -- set the number of core available for experimentation" + "\n\t\t"
171-
+ SOCKET_PARAMETER + " [socketPort] -- start socket pipeline to interact with another framework"
176+
+ SOCKET_PARAMETER + " [socketPort] -- starts socket pipeline to interact with another framework"
172177
+ "\n\t\t" + TUNNELING_PARAMETER
173178
+ " -- start pipeline to interact with another framework"
179+
+ "\n\t\t" + PING_INTERVAL + " [pingInterval] \t\t "
180+
+ "-- when in server mode (socket parameter set), defines in milliseconds the time "
181+
+ "between each ping packet sent to clients to keep alive the connection. "
182+
+ "The default value is 10000, set to -1 to deactivate this behaviour."
174183
+ "\n\t=== Infos ===" + "\n\t\t" + HELP_PARAMETER
175184
+ " -- get the help of the command line" + "\n\t\t" + GAMA_VERSION
176185
+ " -- get the the version of gama" + "\n\t=== Library Runner ===" + "\n\t\t"
@@ -234,6 +243,10 @@ private boolean checkParameters(final List<String> args) {
234243
mustContainOutFolder = mustContainInFile = false;
235244
this.socket = Integer.parseInt(after(args, SSOCKET_PARAMETER));
236245
}
246+
if (args.contains(PING_INTERVAL)) {
247+
size = size - 2;
248+
this.ping_interval = Integer.parseInt(after(args, PING_INTERVAL));
249+
}
237250
if (args.contains(THREAD_PARAMETER)) {
238251
size = size - 2;
239252
processorQueue.setNumberOfThreads(Integer.parseInt(after(args, THREAD_PARAMETER)));
@@ -363,13 +376,13 @@ public Object start(final IApplicationContext context) throws Exception {
363376
buildXML(args);
364377
} else if (args.contains(SOCKET_PARAMETER)) {
365378
// GamaListener.newInstance(this.socket, this);
366-
new GamaListener(this.socket, this, false, "", "", "");
379+
new GamaListener(this.socket, this, false, "", "", "", this.ping_interval);
367380
} else if (args.contains(SSOCKET_PARAMETER)) {
368381
final String jks = args.contains(SSOCKET_PARAMETER_JKSPATH) ? after(args, SSOCKET_PARAMETER_JKSPATH) : "";
369382
final String spwd = args.contains(SSOCKET_PARAMETER_SPWD) ? after(args, SSOCKET_PARAMETER_SPWD) : "";
370383
final String kpwd = args.contains(SSOCKET_PARAMETER_KPWD) ? after(args, SSOCKET_PARAMETER_KPWD) : "";
371384
// System.out.println(jks+" "+spwd+" "+kpwd);
372-
new GamaListener(this.socket, this, true, jks, spwd, kpwd);
385+
new GamaListener(this.socket, this, true, jks, spwd, kpwd, ping_interval);
373386
} else {
374387
runSimulation(args);
375388
}

0 commit comments

Comments
 (0)