JobBarrier.java
/*
* Licensed to The Apereo Foundation under one or more contributor license
* agreements. See the NOTICE file distributed with this work for additional
* information regarding copyright ownership.
*
*
* The Apereo Foundation licenses this file to you under the Educational
* Community License, Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain a copy of the License
* at:
*
* http://opensource.org/licenses/ecl2.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*
*/
package org.opencastproject.job.api;
import org.opencastproject.job.api.Job.Status;
import org.opencastproject.serviceregistry.api.ServiceRegistry;
import org.opencastproject.serviceregistry.api.ServiceRegistryException;
import org.opencastproject.util.JobCanceledException;
import org.opencastproject.util.NotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
/**
* This class is a utility implementation that will wait for all given jobs to change their status to either one of:
* <ul>
* <li>{@link Job.Status#FINISHED}</li>
* <li>{@link Job.Status#FAILED}</li>
* <li>{@link Job.Status#DELETED}</li>
* </ul>
*/
public final class JobBarrier {
/** The logging facility */
private static final Logger logger = LoggerFactory.getLogger(JobBarrier.class);
/** Default polling interval is 5 seconds */
public static final long DEFAULT_POLLING_INTERVAL = 5000L;
/** The service registry used to do the polling */
private final ServiceRegistry serviceRegistry;
/** Time in milliseconds between two pools for the job status */
private final long pollingInterval;
/** The job that's waiting */
private final Optional<Long> waiterJobId;
/** The jobs to wait on */
private final List<Job> jobs;
/** An exception that might have been thrown while polling */
private volatile Throwable pollingException = null;
/** The status map */
private volatile Result status = null;
/**
* Creates a barrier without any jobs, using <code>registry</code> to poll for the outcome of the monitored jobs using
* the default polling interval {@link #DEFAULT_POLLING_INTERVAL}. The <code>waiter</code> is the job which is waiting
* for the other jobs to finish. Use {@link #addJob(Job)} to add jobs to monitor.
*
* @param registry
* the registry
* @param waiter
* the job waiting for the other jobs to finish
*/
public JobBarrier(Job waiter, ServiceRegistry registry) {
this(waiter, registry, DEFAULT_POLLING_INTERVAL, new Job[] {});
}
/**
* Creates a barrier for <code>jobs</code>, using <code>registry</code> to poll for the outcome of the monitored jobs
* using the default polling interval {@link #DEFAULT_POLLING_INTERVAL}. The <code>waiter</code> is the job which is
* waiting for the other jobs to finish.
*
* @param registry
* the registry
* @param jobs
* the jobs to monitor
* @param waiter
* the job waiting for the other jobs to finish
*/
public JobBarrier(Job waiter, ServiceRegistry registry, Job... jobs) {
this(waiter, registry, DEFAULT_POLLING_INTERVAL, jobs);
}
/**
* Creates a wrapper for <code>job</code>, using <code>registry</code> to poll for the job outcome. The
* <code>waiter</code> is the job which is waiting for the other jobs to finish.
*
* @param registry
* the registry
* @param pollingInterval
* the time in miliseconds between two polling operations
* @param waiter
* the job waiting for the other jobs to finish
*/
public JobBarrier(Job waiter, ServiceRegistry registry, long pollingInterval) {
this(waiter, registry, pollingInterval, new Job[] {});
}
/**
* Creates a wrapper for <code>job</code>, using <code>registry</code> to poll for the job outcome. The
* <code>waiter</code> is the job which is waiting for the other jobs to finish.
*
* @param jobs
* the job to poll
* @param registry
* the registry
* @param pollingInterval
* the time in miliseconds between two polling operations
* @param waiter
* the job waiting for the other jobs to finish
*/
public JobBarrier(Job waiter, ServiceRegistry registry, long pollingInterval, Job... jobs) {
if (registry == null)
throw new IllegalArgumentException("Service registry must not be null");
if (jobs == null)
throw new IllegalArgumentException("Jobs must not be null");
if (pollingInterval < 0)
throw new IllegalArgumentException("Polling interval must be a positive number");
this.serviceRegistry = registry;
this.pollingInterval = pollingInterval;
if (waiter != null)
this.waiterJobId = Optional.of(waiter.getId());
else
this.waiterJobId = Optional.empty();
this.jobs = new ArrayList<Job>(Arrays.asList(jobs));
}
private void suspendWaiterJob() {
if (this.waiterJobId.isPresent()) {
try {
final Job waiter = serviceRegistry.getJob(waiterJobId.get());
waiter.setStatus(Job.Status.WAITING);
logger.debug("Job {} set to WAITING state.", waiter.getId());
this.serviceRegistry.updateJob(waiter);
} catch (ServiceRegistryException e) {
logger.warn("Unable to put {} into a waiting state, this may cause a deadlock: {}", waiterJobId, e.getMessage());
} catch (NotFoundException e) {
logger.warn("Unable to put {} into a waiting state, job not found by the service registry. This may cause a deadlock: {}", waiterJobId, e.getMessage());
}
} else {
logger.debug("No waiting job set, unable to put waiting job into waiting state");
}
}
private void wakeWaiterJob() {
if (this.waiterJobId.isPresent()) {
try {
final Job waiter = serviceRegistry.getJob(waiterJobId.get());
waiter.setStatus(Job.Status.RUNNING);
logger.debug("Job {} wakened and set back to RUNNING state.", waiter.getId());
this.serviceRegistry.updateJob(waiter);
} catch (ServiceRegistryException e) {
logger.warn("Unable to put {} into a waiting state, this may cause a deadlock: {}", waiterJobId, e.getMessage());
} catch (NotFoundException e) {
logger.warn("Unable to put {} into a waiting state, job not found by the service registry. This may cause a deadlock: {}", waiterJobId, e.getMessage());
}
} else {
logger.debug("No waiting job set, unable to put waiting job into waiting state");
}
}
/**
* Waits for a status change and returns the new status.
*
* @return the status
*/
public Result waitForJobs() {
return waitForJobs(0);
}
/**
* Waits for a status change on all jobs and returns. If waiting for the status exceeds a certain limit, the method
* returns even if some or all of the jobs are not yet finished. The same is true if at least one of the jobs fails or
* gets stopped or deleted.
*
* @param timeout
* the maximum amount of time to wait
* @throws IllegalStateException
* if there are no jobs to wait for
* @throws JobCanceledException
* if one of the jobs was canceled
*/
public Result waitForJobs(long timeout) throws JobCanceledException, IllegalStateException {
if (jobs.size() == 0)
return new Result(new HashMap<Job, Status>());
this.suspendWaiterJob();
synchronized (this) {
JobStatusUpdater updater = new JobStatusUpdater(timeout);
try {
updater.start();
wait();
} catch (InterruptedException e) {
logger.debug("Interrupted while waiting for job");
}
}
if (pollingException != null) {
if (pollingException instanceof JobCanceledException)
throw (JobCanceledException) pollingException;
throw new IllegalStateException(pollingException);
}
this.wakeWaiterJob();
return getStatus();
}
/**
* Adds the job to the list of jobs to wait for. An {@link IllegalStateException} is thrown if the barrier has already
* been asked to wait for jobs by calling {@link #waitForJobs()}.
*
* @param job
* the job
* @throws IllegalStateException
* if the barrier already started waiting
*/
public void addJob(Job job) throws IllegalStateException {
if (job == null)
throw new IllegalArgumentException("Job must not be null");
jobs.add(job);
}
/**
* Sets the outcome of the various jobs that were monitored.
*
* @param status
* the status
*/
void setStatus(Result status) {
this.status = status;
}
/**
* Returns the resulting status map.
*
* @return the status of the individual jobs
*/
public Result getStatus() {
return status;
}
/** Thread that keeps polling for status changes. */
class JobStatusUpdater extends Thread {
/** Maximum wait in milliseconds or 0 for unlimited waiting */
private final long workTime;
/**
* Creates a new status updater that will wait for finished jobs. If <code>0</code> is passed in as the work time,
* the updater will wait as long as it takes. Otherwise, it will stop after the indicated amount of time has passed.
*
* @param workTime
* the work time
*/
JobStatusUpdater(long workTime) {
this.workTime = workTime;
}
@Override
public void run() {
final long endTime = workTime > 0 ? System.currentTimeMillis() + workTime : 0;
final Map<Job, Job.Status> finishedJobs = new HashMap<Job, Job.Status>();
while (true) {
final long time = System.currentTimeMillis();
// Wait a little..
try {
final long timeToSleep = Math.min(pollingInterval, Math.abs(endTime - time));
Thread.sleep(timeToSleep);
} catch (InterruptedException e) {
logger.debug("Job polling thread was interrupted");
return;
}
// Look at all jobs and make sure all of them have reached the expected status
for (final Job job : jobs) {
// Don't ask if we already know
if (!finishedJobs.containsKey(job)) {
// Get the job status from the service registry
try {
final Job processedJob = serviceRegistry.getJob(job.getId());
final Job.Status jobStatus = processedJob.getStatus();
switch (jobStatus) {
case CANCELLED:
throw new JobCanceledException(processedJob);
case DELETED:
case FAILED:
case FINISHED:
job.setStatus(jobStatus);
job.setPayload(processedJob.getPayload());
finishedJobs.put(job, jobStatus);
break;
case PAUSED:
case QUEUED:
case RESTART:
case DISPATCHING:
case INSTANTIATED:
case RUNNING:
logger.trace("{} is still in the works", job);
break;
case WAITING:
logger.trace("{} is waiting", job);
break;
default:
logger.error("Unhandled job status '{}' found", jobStatus);
break;
}
} catch (NotFoundException e) {
logger.warn("Error polling job {}: Not found!", job);
finishedJobs.put(job, Job.Status.DELETED);
pollingException = e;
break;
} catch (ServiceRegistryException e) {
logger.warn("Error polling service registry for the status of {}: {}", job, e.getMessage());
} catch (JobCanceledException e) {
logger.warn("Job {} got canceled", job);
pollingException = e;
updateAndNotify(finishedJobs);
return;
} catch (Throwable t) {
logger.error("An unexpected error occured while waiting for jobs", t);
pollingException = t;
updateAndNotify(finishedJobs);
return;
}
}
// Are we done already?
if (finishedJobs.size() == jobs.size()) {
updateAndNotify(finishedJobs);
return;
} else if (workTime > 0 && endTime >= time) {
pollingException = new InterruptedException("Timeout waiting for job processing");
updateAndNotify(finishedJobs);
return;
}
}
}
}
/**
* Notifies listeners about the status change.
*
* @param status
* the status
*/
private void updateAndNotify(Map<Job, Job.Status> status) {
JobBarrier.this.setStatus(new Result(status));
synchronized (JobBarrier.this) {
JobBarrier.this.notifyAll();
}
}
}
/** Result of a waiting operation on a certain number of jobs. */
public static class Result {
/** The outcome of this barrier */
private final Map<Job, Job.Status> status;
/**
* Creates a new job barrier result.
*
* @param status
* the barrier outcome
*/
public Result(Map<Job, Job.Status> status) {
this.status = status;
}
/**
* Returns the status details.
*
* @return the status details
*/
public Map<Job, Job.Status> getStatus() {
return status;
}
/**
* Returns <code>true</code> if all jobs are in the <code>{@link Job.Status#FINISHED}</code> state.
*
* @return <code>true</code> if all jobs are finished
*/
public boolean isSuccess() {
for (final Job.Status state : status.values()) {
if (!state.equals(Job.Status.FINISHED))
return false;
}
return true;
}
}
}