/*
 * RiskScape™ Copyright New Zealand Institute for Earth Science Limited
 * (Earth Sciences New Zealand) is distributed for research purposes only
 * under the terms of AGPLv3.
 *
 * RiskScape™ Copyright 2026 New Zealand Institute for Earth Science
 * Limited (Earth Sciences New Zealand). All rights reserved. Source code
 * available under the AGPLv3.
 * 
 * This program is free software: you can redistribute it and/or modify it under
 *  the terms of the GNU Affero General Public License as published by the Free
 *  Software Foundation, either version 3 of the License, or (at your option) any
 *  later version.
 * 
 * This program is distributed for RESEARCH PURPOSES ONLY, in the hope that it will
 * be useful for research and education initiatives.
 * 
 * If you are not a researcher, or you are a researcher who wishes to use this
 * program on terms other than AGPLv3 (including those who wish to restrict the
 * distribution of any source code created using this program), please contact:
 * https://riskscape.org.nz
 * 
 * This program is distributed WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Affero General Public License for more details.  You should have received a copy
 * of the GNU Affero General Public License along with this program.  If not, see
 * <http://www.gnu.org/licenses/>.
 * 
 * By way of summary only, under the AGPLv3:
 *     • Permissions of this strongest copyleft license are conditioned
 *       on making available complete source code of licensed works and
 *       modifications, which include larger works using a licensed work,
 *       under the same license.
 *     • Copyright and license notices must be preserved.
 *     • Contributors provide an express grant of patent rights.
 *     • When a modified version is used to provide a service over a
 *       network, the complete source code of the modified version must be made
 *       available.
 */
package nz.org.riskscape.engine.sched;

import java.lang.Thread.State;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Metered;
import com.codahale.metrics.Metric;
import com.codahale.metrics.MetricRegistry;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import nz.org.riskscape.engine.RiskscapeException;
import nz.org.riskscape.engine.output.PipelineJobContext;
import nz.org.riskscape.engine.pipeline.ExecutionResult;
import nz.org.riskscape.engine.problem.ProblemFactory;
import nz.org.riskscape.engine.sched.Worker.Result;
import nz.org.riskscape.engine.task.ReturnState;
import nz.org.riskscape.engine.task.TaskSpec;
import nz.org.riskscape.engine.task.TaskState;
import nz.org.riskscape.engine.task.WorkerTask;
import nz.org.riskscape.engine.task.WritePageBuffer;
import nz.org.riskscape.problem.Problem;
import nz.org.riskscape.problem.ProblemException;
import nz.org.riskscape.problem.Problems;

/**
 * The Scheduler oversees the execution of the pipeline tasks. Mostly this
 * involves assigning WorkerTasks to the Worker threads.
 *
 * TODO rename methods called by scheduler thread to make it clearer
 */
@Slf4j
public class Scheduler implements Runnable {

  public interface LocalProblems extends ProblemFactory {
    static LocalProblems get() {
      return Problems.get(LocalProblems.class);
    }

    // a normally-emitting part of the pipeline accepted tuples, but did not produce any
    Problem noDataProduced(String from);

    // scheduler thread uncaught exception - scheduler is dead now
    Problem crash(String exceptionMessage);

    // scheduler has waiting tasks, but none of them can run
    Problem deadlock();

    // scheduler is stopped - an existing or a new job can return this problem to say it can't run
    Problem stopped();
  }

  private static final WorkerTask NO_TASKS_WAITING = new WorkerTask() {
    @Override
    public ReturnState run() {
      return null;
    }

    @Override
    public boolean producesResult() {
      return false;
    }
  };

  private static final WorkerTask NO_TASKS_READY = new WorkerTask() {
    @Override
    public ReturnState run() {
      return null;
    }

    @Override
    public boolean producesResult() {
       return false;
    }
  };

  // don't sleep indefinitely, just in case of race conditions
  public static final int MAX_WAIT_MILLISECS = 100;

  private List<Worker> workers = new ArrayList<>();
  private List<Thread> workerThreads = new ArrayList<>();

  @Getter
  private List<WorkerTask> waitingTasks = new ArrayList<>();
  @Getter
  private List<WorkerTask> runningTasks = new ArrayList<>();
  @Getter
  private List<WorkerTask> completedTasks = new ArrayList<>();

  private Object sleepMutex = new Object();

  private final Thread schedulerThread = new Thread(this, "scheduler-thread");

  /**
   * Flag to say whether the scheduler has been stopped yet
   */
  private boolean stopped = false;

  boolean detectDeadlocks = true;
  private boolean workerFinished = false;
  // indicates if a complete taskSpec has already been found that producted no output. Used to
  // warn on the first occurence only.
  private boolean taskProducedNoData = true;

  private final SchedulerParams params;

  // nb this list can be accessed by threads other than the scheduler thread, so make sure you have the sleepMutex
  // before accessing (including the scheduler thread)
  private LinkedList<ExecutionFuture> jobQueue = new LinkedList<>();

  //  read and written by scheduler thread only, no sync required
  private ExecutionFuture currentJob;

  public Scheduler(SchedulerParams params) {
    this.params = params;

    // create the scheduler's worker threads
    for (int id = 1; id <= params.getNumThreads(); id++) {
      Worker worker = new Worker(id, this);
      workers.add(worker);
      Thread thread = new Thread(worker);
      thread.setDaemon(true);
      thread.setName(String.format("execution-worker-%d", id));
      workerThreads.add(thread);
    }
  }

  void startWorkers() {
    workerThreads.forEach(Thread::start);
  }

  private void stopWorkers() {
    workers.forEach(Worker::stop);
  }

  private void checkForDeadlock(WorkerTask nextTask, long numWorkersRunning) {
    if (detectDeadlocks && nextTask == NO_TASKS_READY && numWorkersRunning == 0) {
      log.error("Deadlock detected!  No worker threads are running and all tasks are blocked");

      Map<TaskSpec, List<WorkerTask>> blockedOnFullOutput = new HashMap<>();
      Map<TaskSpec, List<WorkerTask>> blockedOnDependencies = new HashMap<>();
      Map<TaskSpec, List<WorkerTask>> blockedOnNoInput = new HashMap<>();
      for (WorkerTask task : waitingTasks) {
        if (task.getPageWriter().map(pr -> pr.isFull()).orElse(false)) {
          blockedOnFullOutput.computeIfAbsent(task.getSpec(), k -> new ArrayList<>()).add(task);
        }

        if (task.getSpec().hasOutstandingDependencies()) {
          blockedOnDependencies.computeIfAbsent(task.getSpec(), k -> new ArrayList<>()).add(task);
        }

        if (task.getPageReader().map(pr -> pr.getPagesRead() != 0 && !pr.hasInput()).orElse(false)) {
          blockedOnNoInput.computeIfAbsent(task.getSpec(), k -> new ArrayList<>()).add(task);
        }
      }

      if (!blockedOnFullOutput.isEmpty()) {
        log.warn("  Blocked on full output:");
        blockedOnFullOutput.forEach((spec, tasks) -> {
          log.warn("    {}", spec);
        });
      }

      if (!blockedOnDependencies.isEmpty()) {
        log.warn("  Blocked on dependencies:");
        blockedOnDependencies.forEach((spec, tasks) -> {
          log.warn("    {} - depending on {}", spec, spec.getDependsOn());
        });
      }

      if (!blockedOnNoInput.isEmpty()) {
        log.warn("  Blocked on no input:");
        blockedOnNoInput.forEach((spec, tasks) -> {
          log.warn("    {} - {} / {} tasks blocked", spec, tasks.size(), spec.getWorkerTasks().size());
        });
      }

      if (!completedTasks.isEmpty()) {
        log.warn("  Completed Tasks:");
        this.completedTasks.stream().map(WorkerTask::getSpec).collect(Collectors.toSet()).forEach(spec -> {
          log.warn("    {}", spec);
        });
      }
      throw new RiskscapeException(LocalProblems.get().deadlock());
    }
  }

  long numWorkersRunning() {
    return workers.stream().filter(p -> p.isRunning()).count();
  }

  /**
   * Try to run a worker task, if possible.
   * @return true if a task was run, false otherwise
   */
  boolean runOnce() {
    long numWorkersRunning = numWorkersRunning();

    if (numWorkersRunning == workers.size()) {
      // all workers are busy, we need to wait until one becomes free
      return false;
    }

    // check if any worker tasks finished and have results that we need to look at
    processWorkerResults();

    WorkerTask toRun = findNextTask();

    if (toRun == NO_TASKS_WAITING || toRun == NO_TASKS_READY) {
      checkForDeadlock(toRun, numWorkersRunning);

      // need to wait until a task becomes unblocked
      return false;
    }

    Worker worker = findFreeWorker(toRun);
    if (worker == null) {
      // should never hit this, as we already checked there was a worker free earlier
      log.error("Only {}/{} workers running, but scheduler couldn't find free worker",
          numWorkersRunning, workers.size());
      return false;
    }

    waitingTasks.remove(toRun);
    runningTasks.add(toRun);
    toRun.markStarted();
    ensureSpecIsStarted(toRun);

    log.debug("Running task {} on {}", toRun, worker);
    worker.runTask(toRun);
    return true;
  }

  /**
   * If required, marks the spec as started and registers metrics
   */
  private void ensureSpecIsStarted(WorkerTask task) {
    TaskSpec spec = task.getSpec();
    if (spec.isCreated()) {
      spec.changeState(TaskState.STARTED);
      MetricRegistry registry = spec.getExecutionContext().getMetricRegistry();
      for (Entry<String, Metric> metricEntry : spec.getMetrics().entrySet()) {
        String name = task.getSpecNameBrief() + "." + metricEntry.getKey();
        registry.register(name, metricEntry.getValue());
      }

      for (Entry<String, Metric> keyMetric : spec.getProgressMetrics().entrySet()) {
        // Add any key progress metrics now the job is starting
        spec.getJobContext().getProgressMetrics().register(
            task.getSpecNameBrief() + "." + keyMetric.getKey(),
            keyMetric.getValue());
      }
    }
  }

  @Override
  public void run() {
    try {
      startWorkers();

      while (!stopped) {

        boolean busy = runOnce();

        synchronized (sleepMutex) {
          busy = busy || !jobQueue.isEmpty();

          if (!busy) {
            waitForNotify();
          }
        }

        // current job is done, let's clean up
        if (currentJob != null && allTasksComplete()) {
          if (currentJob != null) {
            currentJob.markComplete();
            cleanupCurrentJob();
          }
        }

        // no current job, see if we can pull one off the queue
        if (currentJob == null) {
          ExecutionFuture jobToProcess;
          synchronized (sleepMutex) {
            jobToProcess = jobQueue.isEmpty() ? null : jobQueue.removeFirst();
          }

          if (jobToProcess != null) {
            prepareNextJob(jobToProcess);
          }
        }
      }

      stopWorkers();
    } catch (Throwable e) {
      // if someone's waiting, we better let them know that their bus has been cancelled
      log.error("Unhandled exception in scheduler, exiting thread loop!", e);
      if (this.currentJob != null) {
        currentJob.markFailed(LocalProblems.get().crash(e.getMessage()).withException(e));
        currentJob = null;
      }

      // let's shut ourselves down properly
      stop(true);
    }
  }

  private List<TaskSpec> buildTasks(ExecutionFuture queuedJob) {
    return new TaskBuilder(params).convertToTasks(queuedJob.getJobContext());
  }

  public boolean allTasksComplete() {
    return waitingTasks.isEmpty() && runningTasks.isEmpty() && completedTasks.size() > 0;
  }

  /**
   * Stop the scheduler from scheduling more work
   * @param join if true, this method will only return once all threads have ended.  This can block if any workers are
   * blocked, but otherwise should return fairly promptly
   */
  public void stop(boolean join) {

    List<ExecutionFuture> toFail;
    synchronized (sleepMutex) {
      stopped = true;

      // in an abundance of caution, we clone the list within the mutex to avoid having to iterate over it where
      // there's a chance of it being modified
      toFail = new ArrayList<>(jobQueue.size());
      toFail.addAll(jobQueue);
      jobQueue.clear();

      // this wakes up the scheduler thread, which will spot the stopped flag and start shutting down the
      // worker threads
      sleepMutex.notify();

    }

    // papa's never comin' home, sniff
    for (ExecutionFuture executionFuture : toFail) {
      executionFuture.markFailed(LocalProblems.get().stopped());
    }

    if (join) {
      // we can return once all the workers have terminated
      workerThreads.forEach(t -> {
        try {
          t.join();
        } catch (InterruptedException e) {
        }
      });
    }
  }

  void processWorkerResults() {
    synchronized (sleepMutex) {
      // clear the flag first (as workers may finish while we're cleaning up)
      workerFinished = false;
    }

    for (Worker worker: workers) {
      if (!worker.isRunning()) {
        processLastTaskResult(worker);
      }
    }
  }

  private int indexOfFirstIdleTask(TaskSpec spec) {
    int index = 0;
    for (WorkerTask task : waitingTasks) {
      if (task.getSpec() == spec && !task.hasPageInProgress()) {
        break;
      }
      index++;
    }
    return index;
  }

  private void queueWork(WorkerTask task) {
    // add the task back to the waiting list. If the task didn't get all the
    // way through the page it was working on, then add it back earlier in
    // the list, so it gets more preference (over idle tasks) next time
    if (task.hasPageInProgress()) {
      waitingTasks.add(indexOfFirstIdleTask(task.getSpec()), task);
    } else {
      waitingTasks.add(task);
    }
  }

  /**
   * Do clean-up after a worker has finished running. We need to check if the task
   * it ran is fully complete, or just blocked waiting on input/output buffers.
   *
   * Called from the scheduler thread so job state can be freely/safely changed here
   */
  private void processLastTaskResult(Worker worker) {
    Result workDone = worker.clearLastTask();

    if (workDone == null) {
      // worker was just idle - there's no cleanup to do
      return;
    }

    WorkerTask lastTask = workDone.task;
    if (! this.runningTasks.contains(lastTask)) {
      // this task shouldn't be running now. Problably due to some other task failing and then
      // cleaning up the job with cleanupCurrentJob()
      // let's just return as we don't want to do any more work on this task.
      return;
    }
    TaskSpec taskSpec = lastTask.getSpec();

    // propagate exceptions and clean up if the worker died mid-task.
    if (workDone.state == ReturnState.ERROR_THROWN) {
      processLastTaskError(workDone);
      return;
    }

    // we're not running any more, that's why we are here
    runningTasks.remove(lastTask);

    log.debug("Finished running task {} with {}", lastTask, workDone.state);

    if (workDone.state == ReturnState.COMPLETE) {

      // see if there's a processing result...
      if (lastTask.producesResult()) {
        Object processingResult = lastTask.consumeProcessingResult();
        if (processingResult == null) {
          throw new NullPointerException("worker was supposed to produce a result " + lastTask);
        }

        waitingTasks.stream().map(WorkerTask::getSpec).collect(Collectors.toSet()).forEach(waitingSpec -> {
          if (waitingSpec.hasDependency(taskSpec)) {
            // give the result to any dependent tasks
            waitingSpec.addProcessingResultFromDependency(taskSpec, processingResult);
          } else if (waitingSpec.hadDependency(taskSpec)) {
            // sanity check / assertion
            throw new IllegalStateException("this should not happen - a worker has completed after a dependent was "
                + "satisifed");
          }
        });
      }

      taskComplete(lastTask);
    } else if (workDone.state != ReturnState.ERROR_THROWN) {
      // the task still has more to do
      queueWork(lastTask);
    }

    // check if the task spec is now fully complete - if there are some tasks that haven't even run and all the others
    // are complete, then we don't need to bother running them - just kill them
    if (taskSpec.allWorkersMatch(wt -> wt.isComplete() || wt.isCreated())) {

      // track how many tuples have been read/written by this task
      long tuplesRead = 0L;
      long tuplesWritten = 0L;
      // Produce a few stats...
      for (WorkerTask task : taskSpec.getWorkerTasks()) {
        tuplesRead += task.getPageReader().map(r -> r.getTuplesRead()).orElse(0L);
        tuplesWritten += task.getPageWriter().map(w -> w.getTuplesWritten()).orElse(1L);

        // don't even bother with these tasks - they've never been run so have no tuples or processing results to share
        if (task.isCreated()) {
          task.markComplete();
          taskComplete(task);
        }
      }

      taskSpec.changeState(TaskState.COMPLETE);

      // clean up on complete
      taskSpec.close();

      if (taskProducedNoData && tuplesRead > 0 && tuplesWritten == 0) {
        // sound the alarm... the task has read tuples but not written any.
        // We only sound the alarm on the first occurence as subsequent occurences
        // could well be caused by this one. For example all data being filtered out before being used
        // as join RHS would otherwise cause multiple warnings.
        taskProducedNoData = false;
        taskSpec.getExecutionContext().getProblemSink().log(
            LocalProblems.get().noDataProduced(taskSpec.getStepsSummary())
                .withSeverity(Problem.Severity.WARNING)
        );
      }

      log.info("Completed step(s) {}", taskSpec.getStepsSummary());
      taskSpec.getOutput().ifPresent(WritePageBuffer::markComplete);

      // clear any dependencies on the completed task, so other things can now run.  Note that at this point all the
      // results from each worker task should have now been given to the dependent tasks via
      // TaskSpec#addResultFromDependency - this bit is about signalling to the scheduler that these tasks should now be
      // ready to run
      // TODO maintain a list of as yet uncompleted specs and run through this instead?
      waitingTasks.stream().map(WorkerTask::getSpec).collect(Collectors.toSet()).forEach(waitingSpec -> {
        if (waitingSpec.hasDependency(taskSpec)) {
          log.info("marking {}'s dependency on {} as satisfied", waitingSpec, taskSpec);
          waitingSpec.satisfyDependency(taskSpec);
        }
      });

      cleanupMetrics(lastTask);
    }
  }

  /**
   * Called by the scheduler thread during processLastTaskResult to handle a failed pipeline
   * @param workDone
   */
  private void processLastTaskError(Result workDone) {
    WorkerTask lastTask = workDone.task;
    // clear it out of the queue, wake up anyone waiting
    taskComplete(lastTask);
    // the job has now failed - anyone waiting will now get the bad news
    // NB not sure whether we want to do this now or after we've waited?
    currentJob.markFailed(workDone.errorThrown);

    log.info("Current job has failed, waiting for other tasks to yield...");
    // now lets hold up the scheduler until everything else has stopped
    synchronized (sleepMutex) {
      while (this.workers.stream().anyMatch(Worker::isRunning)) {
        try {
          sleepMutex.wait(0);
        } catch (InterruptedException e) {}
      }
    }

    log.info("All job's tasks have stopped, cleaning up after failure");

    // ok, nothing is running any more, let's clean up
    cleanupCurrentJob();
  }

  /**
   * Called only by the scheduler thread to reset state before potentially running the next job
   */
  private void cleanupCurrentJob() {
    // let these fellows clean up after themselves - we might be being called after a failure
    runningTasks.forEach(this::closeTaskAndSwallowErrors);
    waitingTasks.forEach(this::closeTaskAndSwallowErrors);

    this.completedTasks.clear();
    this.waitingTasks.clear();
    this.runningTasks.clear();

    this.currentJob = null;
  }

  private void closeTaskAndSwallowErrors(WorkerTask task) {
    try {
      task.close();
    } catch (Throwable e) {
      log.error("Exception closing task {}", task, e);
    }
  }

  private void cleanupMetrics(WorkerTask task) {
    TaskSpec taskSpec = task.getSpec();
    MetricRegistry registry = taskSpec.getExecutionContext().getMetricRegistry();
    for (Entry<String, Metric> metricEntry : taskSpec.getMetrics().entrySet()) {
      Metric metric = metricEntry.getValue();

      if (metric instanceof Metered) {
        // meters will keep ticking, which will bring down the average.
        // take a snapshot of the average now, as it will better reflect reality
        Metered meter = (Metered) metric;
        final double average = meter.getOneMinuteRate();
        String name = task.getSpecNameBrief() + "." + metricEntry.getKey();
        registry.register(name + ".per-sec", (Gauge<Double>) () -> average);
      }
    }

    // remove any key progress metrics now the job is stopping
    for (Entry<String, Metric> keyMetric : taskSpec.getProgressMetrics().entrySet()) {
      taskSpec.getJobContext().getProgressMetrics().remove(task.getSpecNameBrief() + "." + keyMetric.getKey());
    }
  }

  private void taskComplete(WorkerTask task) {
    task.markComplete();
    if (!completedTasks.contains(task)) {
      waitingTasks.remove(task);
      completedTasks.add(task);

      try {
        task.close();
      } catch (RuntimeException e) {
        log.error("Exception closing task {} ", task, e);
      }
    }
  }


  private void waitForNotify() {
    synchronized (sleepMutex) {
      // one final sanity-check in case a finished worker sent a mutex.notify()
      // just before we got to the mutex.wait()
      if (workerFinished) {
        return;
      }

      try {
        // if all workers are busy then we should get notified the next time there's something to do.
        // If we've still got free workers, keep polling frequently as a running task may have freed up
        // input/output so that a waiting task can now run (we don't get notified of this)
        int delay = numWorkersRunning() == workers.size() ? MAX_WAIT_MILLISECS : 1;
        sleepMutex.wait(delay);
      } catch (InterruptedException e) {
        throw new RuntimeException("Scheduler interrupted", e);
      }
    }
  }

  /**
   * Notifies the scheduler that a task has finished running for now. This doesn't
   * necessarily mean the task is complete - it may have stopped because it's
   * blocked waiting on input or output. But the worker thread is now idle and can
   * start processing another task.
   */
  public void workerFinished() {
    synchronized (sleepMutex) {
      workerFinished = true;
      sleepMutex.notify();
    }
  }

  private Worker findFreeWorker(WorkerTask task) {
    // find the first idle processor (making sure we've cleaned up
    // any previous task for it already)
    for (Worker worker: workers) {
      if (!worker.isRunning() && worker.lastResult.get() == null) {
        return worker;
      }
    }
    return null;
  }

  private WorkerTask findNextTask() {
    for (WorkerTask task : waitingTasks) {
      if (task.isReadyToRun()) {
        return task;
      }
    }

    return waitingTasks.isEmpty() ? NO_TASKS_WAITING : NO_TASKS_READY;
  }

  /**
   * Adds a list of task specifications to the Scheduler.
   * @return a list of new worker tasks created
   * @throws ProblemException
   */
  List<WorkerTask> addTasks(List<TaskSpec> taskSpecs) throws ProblemException {
    List<WorkerTask> newTasks = new ArrayList<>(taskSpecs.size());

    List<Problem> failures = new LinkedList<>();

    for (TaskSpec spec : taskSpecs) {
      int threadsPerTask = spec.isParallelizable() ? params.getMaxThreadsPerTask() : 1;

      parallelTasks:
      for (int i = 0; i < threadsPerTask; i++) {
        try {
          WorkerTask task = spec.newWorkerTask(this);
          newTasks.add(task);
          // find all the failures, not just the first - we want to give the user as much chance as possible
          // to fix up their problems in one hit
        } catch (ProblemException e) {
          // We use ProblemException here instead of returning ResultOrProblems (somewhat dubiously, as ProblemException
          // is supposed to be for local flow control, to my eternal shame)
          failures.addAll(e.getProblems());
          // don't bother trying to create more tasks for this spec - it's going to fail
          break parallelTasks;
        } catch (Throwable e) {
          // we don't want to crash the scheduler because of buggy output code, catch anything unhandled and report an
          // error back for these too
          failures.add(Problems.caught(e));
        }
      }
    }

    // not all tasks could be created, fail execution
    if (failures.size() > 0) {
      throw new ProblemException(failures);
    }

    this.waitingTasks.addAll(newTasks);
    return newTasks;
  }

  public Future<ExecutionResult> queueJob(PipelineJobContext context) {
    ExecutionFuture queuedJob = new ExecutionFuture(context);

    synchronized (sleepMutex) {

      if (stopped) {
        throw new RiskscapeException(LocalProblems.get().stopped());
      }

      // start the thread if it hasn't already
      if (schedulerThread.getState() == State.NEW) {
        schedulerThread.setDaemon(true);
        schedulerThread.start();
      }

      jobQueue.add(queuedJob);

      sleepMutex.notify();
    }


    return queuedJob;
  }

  /**
   * Called only by the scheduler thread
   *
   * Turns the pipeline job in to tasks for execution, adds them to the various queues ready for the scheduler to start
   * scheduling.
   */
  void prepareNextJob(ExecutionFuture newJob) {
    this.currentJob = newJob;
    // build a list of tasks to run for the job
    List<TaskSpec> newWork = buildTasks(currentJob);
    // add the work to the Scheduler's queue of waiting tasks
    List<WorkerTask> newTasks;
    try {
      newTasks = addTasks(newWork);
    } catch (ProblemException e) {
      log.info("Failed to convert pipeline in to tasks, failing pipeline execution", e);
      currentJob.markFailed(e);
      cleanupCurrentJob();
      return;
    }


    // special case - we've been given an empty pipeline (why?)
    if (newTasks.size() == 0) {
      currentJob.markComplete();
      cleanupCurrentJob();
    }

    log.info("Added {} new tasks", newTasks.size());
    if (log.isDebugEnabled()) {
      int ctr = 0;
      for (WorkerTask task : newTasks) {
        log.debug("  {}: {}", ctr++, task);
      }
    }
  }
}
