affine/jobserv
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'raceConditions' into 'CICD'

Browse source 
Race conditions fixed, all tests in CI pass

See merge request whom/jobserv!5
This commit is contained in:
Aidan Hahn 2019-06-01 22:10:52 +00:00
commit e74c74b909
3 changed files with 66 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

38
src/main/java/JobServ/ProcessController.java
View file

@ -8,11 +8,14 @@
package JobServ;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
/*
* ProcessController
@ -33,18 +36,23 @@ class ProcessController {
private BufferedReader reader;
private Process process;
private Boolean killedManually = false;
private Lock lock;
private int lockTimeout; // seconds
/*
* Constructor
* Takes a command and spawns it in a new process
* Redirects IO streams and assigns a fake PID
*/
public ProcessController(String command) throws IOException {
public ProcessController(String command, int lockTimeout) throws IOException {
this.pid = ProcessController.nextPid;
ProcessController.nextPid += 1;
this.lock = new ReentrantLock();
this.lockTimeout = lockTimeout;
this.process = Runtime.getRuntime().exec(command);
this.output = this.process.getOutputStream();
this.input = this.process.getInputStream();
@ -54,6 +62,28 @@ class ProcessController {
JobServServer.logger.write("Job " + String.valueOf(this.pid) + ": " + command);
}
/*
* getLock()
* attempts to get the lock for lockTimeout seconds
* or throws exceptions if interrupted
*/
public boolean getLock() throws InterruptedException {
return this.lock.tryLock(this.lockTimeout, TimeUnit.SECONDS);
}
/*
* releaseLock()
* releases lock on process
*/
public void releaseLock() {
try {
this.lock.unlock();
} catch (IllegalMonitorStateException e) {
JobServServer.logger.write("Thread tried to release a lock it didnt have! " + e.getMessage());
}
}
/*
* getPid()
* returns translated pid of this process

92
src/main/java/JobServ/ProcessManager.java
View file

@ -40,7 +40,6 @@ class ProcessManager {
* processMap
*/
protected ConcurrentHashMap<Integer, ProcessController> processMap;
protected ConcurrentHashMap<Integer, Boolean> lockMap;
private ExecutorService threadPool = Executors.newCachedThreadPool();
/*
@ -49,7 +48,6 @@ class ProcessManager {
*/
public ProcessManager() {
processMap = new ConcurrentHashMap<Integer, ProcessController>();
lockMap = new ConcurrentHashMap<Integer, Boolean>();
/* TODO: In a long running server over a large period of time
* It is possible that the streams used to redirect IO in the
* Processes may become a significant use of resources.
@ -68,11 +66,9 @@ class ProcessManager {
public int newProcess(String command) {
try {
ProcessController newProc = new ProcessController(command);
ProcessController newProc = new ProcessController(command, this.LOCK_TIMEOUT);
// we dont need to lock the map yet
this.lockMap.put(newProc.getPid(), true);
this.processMap.put(newProc.getPid(), newProc);
this.releaseLock(newProc.getPid());
return newProc.getPid();
@ -95,14 +91,13 @@ class ProcessManager {
public int getProcessStatus(int pid) {
try {
if(!this.getLock(pid)) {
return 3;
// lock could not be grabbed before timeout
JobServServer.logger.write("Timeout getting process status: " + String.valueOf(pid));
return 4;
}
} catch (TimeoutException e) {
// lock could not be grabbed before timeout
JobServServer.logger.write("Timeout getting process " +
String.valueOf(pid) + " status: " + e.getMessage());
return 4;
} catch (IndexOutOfBoundsException e) {
return 3;
}
ProcessController candidate = this.processMap.get(pid);
@ -123,13 +118,12 @@ class ProcessManager {
public int getProcessReturn(int pid) {
try {
if(!this.getLock(pid)) {
return 258;
JobServServer.logger.write("Timeout getting process return: " + String.valueOf(pid));
return 259;
}
} catch (TimeoutException e) {
JobServServer.logger.write("Timeout getting process " +
String.valueOf(pid) + " return: " + e.getMessage());
return 259;
} catch (IndexOutOfBoundsException e) {
return 258;
}
ProcessController candidate = this.processMap.get(pid);
@ -146,13 +140,12 @@ class ProcessManager {
public String getProcessOutput(int pid, int lines) {
try {
if(!this.getLock(pid)) {
return "[-] SERVER: Process not found";
JobServServer.logger.write("Timeout getting process output: " + String.valueOf(pid));
return "[-] SERVER: Timeout grabbing lock to access process information";
}
} catch (TimeoutException e) {
JobServServer.logger.write("Timeout getting process " +
String.valueOf(pid) + " output: " + e.getMessage());
return "[-] SERVER: Timeout grabbing lock to access process information";
} catch (IndexOutOfBoundsException e) {
return "[-] SERVER: Process not found";
}
ProcessController candidate = this.processMap.get(pid);
@ -172,13 +165,13 @@ class ProcessManager {
public int killProcess(int pid) {
try {
if(!this.getLock(pid)) {
return 2;
JobServServer.logger.write("Timeout killing process: " + String.valueOf(pid));
return 3;
}
} catch (TimeoutException e) {
JobServServer.logger.write("Timeout killing process " +
String.valueOf(pid) + ": " + e.getMessage());
return 3;
} catch (IndexOutOfBoundsException e) {
return 2;
}
ProcessController candidate = this.processMap.get(pid);
@ -194,45 +187,20 @@ class ProcessManager {
* Function is synchronized to prevent multiple threads accessing the same lock at once
* (ConcurrentHashMap will report whatever lock value was last to successfully update)
*/
protected synchronized Boolean getLock(int pid) throws TimeoutException {
if (!lockMap.containsKey(pid)) {
return false;
protected synchronized Boolean getLock(int pid) throws IndexOutOfBoundsException {
ProcessController candidate = this.processMap.get(pid);
if (candidate == null) {
throw new IndexOutOfBoundsException();
}
Future<Object> future = this.threadPool.submit(
new Callable<Object>() {
public Object call() {
while(lockMap.get(pid)) {
continue; // spin!
}
lockMap.replace(pid, true);
return 1;
}
});
try {
future.get(this.LOCK_TIMEOUT, TimeUnit.SECONDS);
Boolean success = candidate.getLock();
return success;
} catch (InterruptedException e) {
JobServServer.logger.write("[!] Couldnt get lock " +
String.valueOf(pid) + ": "+ e.getMessage());
future.cancel(true);
// in case lock was grabbed after exception
this.releaseLock(pid);
return false;
} catch (ExecutionException e) {
JobServServer.logger.write("[!] Couldnt get lock " +
String.valueOf(pid) + ": "+ e.getMessage());
future.cancel(true);
// in case lock was grabbed after exception
this.releaseLock(pid);
return false;
// cancel the attempt to grab the lock
}
/*
@ -247,8 +215,6 @@ class ProcessManager {
* mediate access to the ProcessManager
* object for fresh calls as well.
*/
return true;
}
/*
@ -256,7 +222,13 @@ class ProcessManager {
* releases mutex so other threads can operate on processqueue
*/
protected void releaseLock(int pid) {
this.lockMap.put(pid, false);
ProcessController candidate = this.processMap.get(pid);
if (candidate == null) {
JobServServer.logger.write("Tried to release lock of process that doesnt exist!");
return;
}
candidate.releaseLock();
}
/*

8
src/test/java/JobServ/ProcessManagerTestImplementation.java
View file

@ -27,18 +27,10 @@ class ProcessManagerTestImplementation extends ProcessManager {
super.releaseLock(pid);
} catch (TimeoutException e) {
System.err.println("[!!] Long Call wasnt able to grab lock!");
return;
} catch (InterruptedException e) {
super.releaseLock(pid); // this doesnt happen, dont cancel this task
System.err.println("[3] Released lock: interrupted");
return;
}
}
public Boolean reportLockState(int pid) {
return super.lockMap.get(pid);
}
}