Commit feaa9e0a authored by Daniel Seybold's avatar Daniel Seybold

some refactoring regarding TSBS runner

parent ef0370f9
Pipeline #80755 passed with stage
in 22 minutes and 24 seconds
......@@ -221,7 +221,7 @@ public class WorkloadApiServiceImpl extends WorkloadApiService {
workloadProcessInstance.setTimestamp(BigDecimal.valueOf(workloadProcess.getStartTimestamp()));
workloadProcessInstance.setProcessId(workloadProcess.getProcessId());
Runner tsbsRunner = new Runner(workloadProcessInstance, workload, taskId);
Runner tsbsRunner = new Runner(workload, taskId);
new Thread(tsbsRunner).start();
return Response.ok().entity(workloadProcessInstance).build();
......
package de.uulm.omi.workload.tsbs;
import de.uulm.omi.cloudiator.dbms.evaluator.model.ProcessStatus;
import de.uulm.omi.cloudiator.dbms.evaluator.model.TSBSWorkload;
import de.uulm.omi.cloudiator.dbms.evaluator.model.WorkloadProcessInstance;
import de.uulm.omi.utils.WorkloadApiProperties;
import de.uulm.omi.workload.WorkloadProcess;
import java.io.IOException;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -29,21 +26,18 @@ public class Runner implements Runnable {
private final int CHECKING_INTERVAL = 5000;
private TSBSWorkload workload;
private WorkloadProcessInstance workloadProcessInstance;
private String taskId;
public Runner(WorkloadProcessInstance workloadProcessInstance, TSBSWorkload workload,
String taskId) {
public Runner(TSBSWorkload workload, String taskId) {
this.workload = workload;
this.workloadProcessInstance = workloadProcessInstance;
this.taskId = taskId;
}
@Override
public void run() {
WorkloadProcess workloadProcess = WorkloadProcess.getInstance();
Properties properties = WorkloadApiProperties.getInstance().getProperties();
Long workloadTimestamp = System.currentTimeMillis();
......@@ -57,16 +51,12 @@ public class Runner implements Runnable {
generateDataCommand += properties.getProperty(WorkloadApiProperties.WORKLOAD_DATA_FOLDER) + workload.getDbms().toString().toLowerCase() + "-data.gz";
LOGGER.debug("Generate data command: " + generateDataCommand);
ProcessBuilder genDataProcess= new ProcessBuilder();
genDataProcess.command(UNIX_SHELL, UNIX_COMMAND_PARAM, generateDataCommand);
Process processData = null;
try {
processData = genDataProcess.start();
} catch (IOException e) {
e.printStackTrace();
}
while (processData.isAlive()) {
workloadProcess
.startWorkload(TSBS_BINARY, UNIX_SHELL, UNIX_COMMAND_PARAM, generateDataCommand, taskId,
workloadTimestamp);
while (workloadProcess.isRunning()) {
try {
LOGGER.debug("Data generation in process...");
Thread.sleep(CHECKING_INTERVAL);
......@@ -75,9 +65,8 @@ public class Runner implements Runnable {
}
}
LOGGER.debug("Phase 1: data generation finished with exit code: " + processData.exitValue());
if(processData.exitValue() != 0){
workloadProcessInstance.setProcessStatus(ProcessStatus.ERROR);
LOGGER.debug("Phase 1: data generation finished with exit code: " + workloadProcess.getExitCode());
if(workloadProcess.getExitCode() != 0){
Thread.currentThread().interrupt();
return;
}
......@@ -89,16 +78,11 @@ public class Runner implements Runnable {
generateQueriesCommand += properties.getProperty(WorkloadApiProperties.WORKLOAD_DATA_FOLDER) + workload.getDbms().toString().toLowerCase() + "-queries.gz" ;
LOGGER.debug(generateQueriesCommand);
workloadProcess
.startWorkload(TSBS_BINARY, UNIX_SHELL, UNIX_COMMAND_PARAM, generateQueriesCommand, taskId,
workloadTimestamp);
ProcessBuilder genQueryProcess = new ProcessBuilder();
genQueryProcess.command(UNIX_SHELL, UNIX_COMMAND_PARAM, generateQueriesCommand);
Process processGenQueries = null;
try {
processGenQueries = genQueryProcess.start();
} catch (IOException e) {
e.printStackTrace();
}
while (processGenQueries.isAlive()) {
while (workloadProcess.isRunning()) {
try {
LOGGER.debug("Query generation in process...");
Thread.sleep(CHECKING_INTERVAL);
......@@ -106,36 +90,38 @@ public class Runner implements Runnable {
e.printStackTrace();
}
}
LOGGER.debug("Phase 2: query generation finished with exit code: " + processGenQueries.exitValue());
if(processGenQueries.exitValue() != 0){
workloadProcessInstance.setProcessStatus(ProcessStatus.ERROR);
LOGGER.debug("Phase 2: query generation finished with exit code: " + workloadProcess.getExitCode());
if(workloadProcess.getExitCode() != 0){
Thread.currentThread().interrupt();
return;
}
//Phase 3: executing insert/load phase
//clear database
String clearDBCommand = TSBSCommandBuilder.clearDB(workload.getDbms(), workload.getTsbsLoadExecution().getDbName(),workload.getDbEndpoints());
ProcessBuilder processClearDBBuilder = new ProcessBuilder();
processClearDBBuilder.command(UNIX_SHELL, UNIX_COMMAND_PARAM, clearDBCommand);
LOGGER.debug(clearDBCommand);
Process clearDBProcess = null;
try {
clearDBProcess = processClearDBBuilder.start();
} catch (IOException e) {
e.printStackTrace();
}
while (clearDBProcess.isAlive()) {
/*
//clear database
String clearDBCommand = TSBSCommandBuilder.clearDB(workload.getDbms(), workload.getTsbsLoadExecution().getDbName(),workload.getDbEndpoints());
ProcessBuilder processClearDBBuilder = new ProcessBuilder();
processClearDBBuilder.command(UNIX_SHELL, UNIX_COMMAND_PARAM, clearDBCommand);
LOGGER.debug(clearDBCommand);
Process clearDBProcess = null;
try {
LOGGER.debug("Clear DB in process...");
Thread.sleep(CHECKING_INTERVAL);
} catch (InterruptedException e) {
clearDBProcess = processClearDBBuilder.start();
} catch (IOException e) {
e.printStackTrace();
}
}
LOGGER.debug("Clear DB finished with exit code: " + clearDBProcess.exitValue());
while (clearDBProcess.isAlive()) {
try {
LOGGER.debug("Clear DB in process...");
Thread.sleep(CHECKING_INTERVAL);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
LOGGER.debug("Clear DB finished with exit code: " + clearDBProcess.exitValue());
*/
......@@ -144,16 +130,11 @@ public class Runner implements Runnable {
loadCommand += " > " + properties.getProperty(WorkloadApiProperties.WORKLOAD_RESULTS_FOLDER) + TSBS_RESULT_FOLDER + taskId.trim() + "_load.txt";
LOGGER.debug(loadCommand);
workloadProcess
.startWorkload(TSBS_BINARY, UNIX_SHELL, UNIX_COMMAND_PARAM, loadCommand, taskId,
workloadTimestamp);
ProcessBuilder loadProcess = new ProcessBuilder();
loadProcess.command(UNIX_SHELL, UNIX_COMMAND_PARAM, loadCommand);
Process processLoad = null;
try {
processLoad = loadProcess.start();
} catch (IOException e) {
e.printStackTrace();
}
while (processLoad.isAlive()) {
while (workloadProcess.isRunning()) {
try {
LOGGER.debug("Load phase in process...");
Thread.sleep(CHECKING_INTERVAL);
......@@ -161,9 +142,8 @@ public class Runner implements Runnable {
e.printStackTrace();
}
}
LOGGER.debug("Phase 3: load phase finished with exit code: " + processLoad.exitValue());
if(processLoad.exitValue() != 0){
workloadProcessInstance.setProcessStatus(ProcessStatus.ERROR);
LOGGER.debug("Phase 3: load phase finished with exit code: " + workloadProcess.getExitCode());
if(workloadProcess.getExitCode() != 0){
Thread.currentThread().interrupt();
return;
}
......@@ -175,9 +155,8 @@ public class Runner implements Runnable {
String executeQueriesCommand = TSBSCommandBuilder.executeQueriesCommand(TSBS_BINARY, properties.getProperty(WorkloadApiProperties.WORKLOAD_DATA_FOLDER), workload.getDbms(), workload.getTsbsQueryExecution(), workload.getDbEndpoints() );
executeQueriesCommand += " > " + properties.getProperty(WorkloadApiProperties.WORKLOAD_RESULTS_FOLDER) + TSBS_RESULT_FOLDER + taskId.trim() + ".txt";
LOGGER.debug(loadCommand);
WorkloadProcess workloadProcess = WorkloadProcess.getInstance();
workloadProcess = WorkloadProcess.getInstance();
workloadProcess
.startWorkload(TSBS_BINARY, UNIX_SHELL, UNIX_COMMAND_PARAM, executeQueriesCommand, taskId,
workloadTimestamp);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment