andi/perfdb
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Activity

Merge remote-tracking branch 'origin/master' into dashboard

Browse Source
This commit is contained in:
Andreas Huber
2023-03-01 18:59:27 +01:00
parent de9218c3c4 be550ebac5
commit 71410801dc
8 changed files with 108 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pdb-ui/build.gradle
View File

@@ -10,6 +10,7 @@ applicationDefaultJvmArgs = [
]
dependencies {
implementation project(':pdb-api')
implementation project(':performanceDb')
implementation project(':pdb-plotting')
implementation project(':pdb-js')

21
pdb-ui/src/main/java/org/lucares/pdbui/PdbController.java
View File

@@ -11,13 +11,13 @@ import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.lucares.pdb.api.AbortException;
import org.lucares.pdb.api.DateTimeRange;
import org.lucares.pdb.api.QueryWithCaretMarker;
import org.lucares.pdb.api.QueryWithCaretMarker.ResultMode;
@@ -32,6 +32,7 @@ import org.lucares.pdbui.domain.PlotResponse;
import org.lucares.pdbui.domain.PlotResponseStats;
import org.lucares.pdbui.job.Job;
import org.lucares.pdbui.job.JobService;
import org.lucares.pdbui.job.JobService.JobEntry;
import org.lucares.performance.db.PerformanceDb;
import org.lucares.recommind.logs.InternalPlottingException;
import org.lucares.recommind.logs.NoDataPointsException;
@@ -104,9 +105,7 @@ public class PdbController implements HardcodedValues, PropertyKeys {
throw new BadRequest("The query must not be empty!");
}
// TODO the UI should cancel requests that are in flight before sending a plot
// request
final Future<ResponseEntity<PlotResponse>> future = jobService
final JobEntry<ResponseEntity<PlotResponse>> jobEntry = jobService
.runJob(new Job<ResponseEntity<PlotResponse>>(request.getSubmitterId()) {
@Override
@@ -140,18 +139,26 @@ public class PdbController implements HardcodedValues, PropertyKeys {
final long deadline = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5);
while (System.currentTimeMillis() < deadline) {
try {
return future.get(1, TimeUnit.SECONDS);
final ResponseEntity<PlotResponse> responseEntity = jobEntry.getFuture().get(1, TimeUnit.SECONDS);
return responseEntity;
} catch (final TimeoutException e) {
LOGGER.info("job for submitter {} still running", request.getSubmitterId());
} catch (final CancellationException e) {
throw new PlotAbortedWebException();
} catch (final ExecutionException e) {
if (e.getCause() instanceof AbortException) {
throw new PlotAbortedWebException();
} else {
LOGGER.error(e.getMessage(), e);
throw new InternalServerError("Internal Server Error", e);
}
}
}
LOGGER.info("job for submitter {} timed out, will cancel the job", request.getSubmitterId());
future.cancel(true);
jobEntry.getJobThreadLocal().cancel();
throw new InternalServerError("Internal Server Error");
} catch (InterruptedException | ExecutionException e) {
} catch (final InterruptedException e) {
LOGGER.error(e.getMessage(), e);
throw new InternalServerError("Internal Server Error", e);
}

8
pdb-ui/src/main/java/org/lucares/pdbui/job/Job.java
View File

@@ -1,9 +1,13 @@
package org.lucares.pdbui.job;
import org.lucares.pdb.api.JobAborter;
public abstract class Job<T> {
private final String submitterId;
private final JobAborter jobAborter = new JobAborter();
public Job(final String submitterId) {
this.submitterId = submitterId;
}
@@ -13,4 +17,8 @@ public abstract class Job<T> {
}
public abstract T executeJob();
public JobAborter getJobAborter() {
return jobAborter;
}
}

64
pdb-ui/src/main/java/org/lucares/pdbui/job/JobService.java
View File

@@ -11,6 +11,8 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.lucares.pdb.api.AbortException;
import org.lucares.pdb.api.JobAborter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -21,23 +23,50 @@ public class JobService implements AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(JobService.class);
private static final int MAX_JOBS = 1;
private static final int MAX_JOBS = 5;
public static class JobEntry<T> {
private final Future<T> future;
private final Job<T> job;
private final JobAborter jobAborter;
public JobEntry(final Future<T> future, final Job<T> job, final JobAborter jobThreadLocal) {
super();
this.future = future;
this.job = job;
this.jobAborter = jobThreadLocal;
}
public Future<T> getFuture() {
return future;
}
public Job<T> getJob() {
return job;
}
public JobAborter getJobThreadLocal() {
return jobAborter;
}
}
private final ExecutorService threadPool = new ThreadPoolExecutor(0, MAX_JOBS, 1L, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1), new CustomizableThreadFactory("jobs"));
private final Map<String, Future<?>> jobs = new ConcurrentHashMap<>();
private final Map<String, JobEntry<?>> jobs = new ConcurrentHashMap<>();
@SuppressWarnings("unchecked")
public <T> Future<T> runJob(final Job<T> job) {
public <T> JobEntry<T> runJob(final Job<T> job) {
return (Future<T>) jobs.compute(job.getSubmitterId(), (submitterId, oldFuture) -> {
return (JobEntry<T>) jobs.compute(job.getSubmitterId(), (submitterId, oldJobEntry) -> {
if (oldFuture != null) {
if (oldJobEntry != null) {
LOGGER.info("cancelling old job");
oldFuture.cancel(true);
oldJobEntry.getJobThreadLocal().cancel();
try {
oldFuture.get(5, TimeUnit.SECONDS);
oldJobEntry.getFuture().get(5, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException e) {
LOGGER.info("old job finished with exception", e);
} catch (final CancellationException e) {
@@ -46,15 +75,18 @@ public class JobService implements AutoCloseable {
LOGGER.info("old job did not finish within 5 seconds");
}
}
final JobAborter jobAborter = job.getJobAborter();
return threadPool.submit(() -> {
final Future<T> future = threadPool.submit(() -> {
try {
AbortException.setAborter(jobAborter);
LOGGER.info("starting job for submitter {}", job.getSubmitterId());
return job.executeJob();
} finally {
jobs.remove(job.getSubmitterId());
}
});
return new JobEntry<T>(future, job, jobAborter);
});
}
@@ -70,15 +102,25 @@ public class JobService implements AutoCloseable {
jobs.computeIfPresent(submitterId, (_submitterId, jobFuture) -> {
LOGGER.info("cancelling old job");
jobFuture.cancel(true);
jobFuture.getJobThreadLocal().cancel();
int waitTime = 0;
try {
jobFuture.get(5, TimeUnit.SECONDS);
final long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5);
while (System.currentTimeMillis() < deadline) {
waitTime = Math.min(waitTime + 2, 500);
jobFuture.getFuture().get(waitTime, TimeUnit.MILLISECONDS);
return null;
}
LOGGER.info("old job did not finish within 5 seconds");
} catch (InterruptedException | ExecutionException e) {
LOGGER.info("old job finished with exception", e);
} catch (final CancellationException e) {
LOGGER.info("cancelled job for submitter {}", submitterId);
} catch (final TimeoutException e) {
LOGGER.info("old job did not finish within 5 seconds");
// LOGGER.info("old job did not finish within 5 seconds");
LOGGER.info("polling wait time was {}", waitTime);
}
return null;