// LocalScheduler.cxx

#include "dial_sched/LocalScheduler.h"
#include <string>
#include <iostream>
#include <fstream>
#include <sstream>
#if ( __GNUC__ == 2 )
  #include "dataset_id/sstream.h"
#else
  #include <sstream>
#endif
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/wait.h>
#include "dataset_util/ssystem.h"
#include "dataset_util/SystemCommand.h"
#include "dataset_util/getcwd.h"
#include "dataset_util/copy_file.h"
#include "dataset_util/FileName.h"
#include "dataset_util/FileStatus.h"
#include "dataset_util/WorkingDirectory.h"
#include "dataset_util/Environment.h"
#include "dataset_util/mkdir.h"
#include "dataset_util/XmlElement.h"
#include "dataset_util/DtdRegistry.h"
#include "dataset_xml/XmlParser.h"
#include "dataset_base/TextDataset.h"
#include "dial_job/JobRepository.h"
#include "dial_job/ChildWatcher.h"
#include "dial_job/JobCreator.h"

using std::string;
using std::cout;
using std::cerr;
using std::ostream;
using std::ofstream;
using std::ostringstream;
using std::endl;
using std::vector;
using std::map;
using std::ifstream;
using std::ostringstream;
using std::auto_ptr;
using dset::Dataset;
using dset::TextDataset;
using dial::Application;
using dial::Task;
using dial::Job;
using dial::JobId;
using dial::JobIdList;
using dial::JobCreator;
using dial::JobRepository;
using dial::ChildWatcher;
using dial::Scheduler;
using dial::LocalScheduler;

typedef LocalScheduler::Name Name;
typedef map<JobId,Job*> JobMap;
typedef map<string,JobId> TaskJobMap;

//**********************************************************************
// Local definitions.
//**********************************************************************

namespace {

//**********************************************************************

// Creator.
Scheduler* create(const XmlElement& ele) {
  if ( ele.name() != LocalScheduler::xml_name() ) return 0;
  if ( ! ele.has_attribute("remove_jobs")) {
    return 0;
  }
  bool remjob = ele.attribute_as_bool("remove_jobs");
  if ( ! ele.has_attribute("jobrep")) {
    return 0;
  }
  string jobrep = ele.attribute("jobrep");
  if ( ele.children().size() != 1 ) return 0;
  const XmlElement* pxjcr = ele.children().front();
  if ( pxjcr == 0 ) return 0;
  // This is a memory leak.
  const JobCreator* pjcr = JobCreator::create(*pxjcr);
  if ( pjcr == 0 ) return 0;
  LocalScheduler* psch = new LocalScheduler(*pjcr, remjob, jobrep);
  return psch;
}

//**********************************************************************

// Register creator.
int STAT_LocalScheduler =
  Scheduler::register_creator(LocalScheduler::xml_name(), create);

//**********************************************************************

// Register the DTD.
DtdRegistry::Status ISTAT_DTD_LocalScheduler =
  DtdRegistry::register_dtd<LocalScheduler>("dial");

//**********************************************************************

// Directory name separator.
string dsep() {
  static string sep = "/";
  return sep;
}

//**********************************************************************

// XML parser.
XmlParser PARSER;

//**********************************************************************

// Return if a directory exists and is readable.

bool check_dir(Name name) {
  if ( name == "" ) return false;
  FileStatus stat(name);
  return stat.is_directory() && stat.is_readable();
}

//**********************************************************************

// Return if a file is regular and is readable.

bool check_file(Name name) {
  if ( name == "" ) return false;
  FileStatus stat(name);
  return stat.is_regular() && stat.is_readable();
}

//**********************************************************************

// Find and check the existence and readability of a directory
// specified by an environmental variable. If succssful, the
// directory name is returned.

Name find_dir(Name ename, const Environment* penv) {
  Name dir;
  if ( penv!=0 && penv->has(ename) ) {
    dir = penv->value(ename);
  } else if ( Environment::initial().has(ename) ) {
    dir = Environment::initial().value(ename);
  }
  // Make the path absolute.
  FileName fname(dir);
  if ( fname.is_relative() ) {
    dir = fname.fullpath().name();
  }
  if ( ! check_dir(dir) ) return "";
  return dir;
}

//**********************************************************************

// Convert int to string.

string int_to_string(int ival) {
  ostringstream osstr;
  osstr << ival;
  return osstr.str();
}

//**********************************************************************

}  // end unnamed namespace

//**********************************************************************
// Implementation class.
//**********************************************************************

class LocalScheduler::Imp {

public:  // data
  const JobCreator& m_cre;
  bool m_valid;
  Name m_appdir;
  Name m_tskdir;
  Name m_jobdir;
  Name m_taskbuilder;
  TaskJobMap m_taskjobs;
  JobIdList m_jids;
  JobMap m_jobs;
  Text m_log;
  bool m_remove_jobs;
  PThreadMutex m_mtx;

  // Job repository.
  string jr_conn;
  JobRepository* pjr;

public:  // functions.

  // Constructor.
  Imp(const JobCreator& cre, bool remove_jobs,
      string jobrep, const Environment* penv =0);

  // Destructor.
  ~Imp();

  // Lock mutex.
  int lock() { return m_mtx.lock(); }

  // Unlock mutex.
  int unlock() { return m_mtx.unlock(); }

  // Message.
  void msg(string txt) const;

};

//**********************************************************************

LocalScheduler::Imp::
Imp(const JobCreator& cre, bool remove_jobs, 
    string jobrep, const Environment* penv0)
: m_cre(cre), m_remove_jobs(remove_jobs), jr_conn(jobrep),
  pjr(new JobRepository()) {
  msg("Creating local scheduler");
  m_valid = false;
  // Select environment.
  const Environment* penv = penv0;
  if ( penv == 0 ) {
    penv = new Environment(Environment::current());
  }
  // Fetch environmental variables.
  m_appdir = find_dir("DIAL_APPS", penv);
  m_tskdir = find_dir("DIAL_TASKS", penv);
  m_jobdir = find_dir("DIAL_JOBS", penv);
  m_taskbuilder = penv->value("DIAL_TASKBUILDER");
  if ( penv0 == 0 ) delete penv;
  if ( m_appdir.size() == 0 ) {
    msg("  DIAL_APPS not defined");
    return;
  }
  if ( m_tskdir.size() == 0 ) {
    msg("  DIAL_TASKS not defined");
    return;
  }
  if ( m_jobdir.size() == 0 ) {
    msg("  DIAL_JOBS not defined");
    return;
  }
  // Define job directory.
  m_jobdir += "/LocalScheduler";
  if ( ! FileStatus(m_jobdir).exists() ) {
    mkdir(m_jobdir);
  }
  // Check apps directory.
  FileStatus appstat = FileStatus(m_appdir);
  msg("  Apps at " + m_appdir);
  if ( ! appstat.is_directory() ||
       ! appstat.is_readable() ||
       ! appstat.is_writeable() ) {
    msg("  Apps directory not found or is not readable and writeable");
    return;
  }
  // Check task directory.
  FileStatus tskstat = FileStatus(m_tskdir);
  msg("  Tasks at " + m_tskdir);
  if ( ! tskstat.is_directory() ||
       ! tskstat.is_readable() ||
       ! tskstat.is_writeable() ) {
    msg( "  Task directory not found or is not readable and writeable");
    return;
  }
  // Check jobs directory.
  FileStatus jobstat = FileStatus(m_jobdir);
  msg("  Jobs at " + m_jobdir);
  if ( ! jobstat.is_directory() ||
       ! jobstat.is_readable() ||
       ! jobstat.is_writeable() ) {
    msg("  Jobs directory not found or is not readable and writeable");
    return;
  }
  m_valid &= FileStatus(m_tskdir).is_directory();
  m_valid &= FileStatus(m_jobdir).is_directory();
  m_valid = true;
  // Job repository.
  if ( jr_conn.size() ) {
    delete pjr;
    pjr = new JobRepository(jr_conn);
    msg("  Creating job repository from");
    msg("    " + jr_conn);
    if ( ! pjr->is_valid() ) {
      msg("  Job repository is invalid!");
    }
  }
}

//**********************************************************************

LocalScheduler::Imp::~Imp() {
  delete pjr;
}

//**********************************************************************

// Log a message.

void LocalScheduler::Imp::msg(string txt) const {
  Text log = const_cast<Text&>(m_log);
  log.append(txt);
  if ( FileStatus("debug_LocalScheduler").exists() ) {
    std::cerr << "LS: " << txt << endl;
  }
}

//**********************************************************************
// Static member functions.
//**********************************************************************

// DTD
// ANY = some kind of job creator

const Text& LocalScheduler::dtd() {
  static Text txt;
  if ( txt.size() == 0 ) {
    txt.append("<!ELEMENT LocalScheduler ANY>");
    txt.append("<!ATTLIST LocalScheduler");
    txt.append("  remove_jobs   CDATA  #REQUIRED");
    txt.append("  jobrep        CDATA  #REQUIRED");
    txt.append(">");
  }
  return txt;
}

//**********************************************************************
// Constructors and destructor.
//**********************************************************************

// Constructor.

LocalScheduler::
LocalScheduler(const JobCreator& cre, bool remove_jobs,
               string jobrep, const Environment* penv)
: m_pimp(new Imp(cre, remove_jobs, jobrep, penv)) {
  if ( penv != 0 ) {
    msg("Scheduler created from current environment");
  } else {
    msg("  Scheduler created from passed environment");
  }
  // Read existing jobs from repository.
  if ( job_repository()->is_valid() ) {
    JobRepository& jr = *job_repository();
    msg("Extracting jobs from JR");
    JobRepository::IdList jids = jr.get_ids();
    msg("  # jobs found: " + int_to_string(jids.size()));
    for ( size_t ijid=0; ijid<jids.size(); ++ijid ) {
      JobId jid = jids[ijid];
      msg("  Job " + jid.to_string());
      const Job* pjob = jr.extract(jid);
      if ( pjob == 0 ) {
        msg("    Unable to extract starting job from JR");
        msg("      " + jr.error_message());
        continue;
      }
      // Following should fail if jobs are not of the expected type.
      Job* pnewjob = cre.convert(*pjob);
      Job& job = *pnewjob;
      if ( ! job.is_valid() ) {
        msg("    Job from JR is invalid");
        continue;
      }
      m_pimp->m_jids.push_back(jid);
      m_pimp->m_jobs[jid] = pnewjob;
      assert( has_job(jid) );
    }
    msg("  # jobs loaded: " + int_to_string(m_pimp->m_jobs.size()));
  }

}

//**********************************************************************

// Destructor.

LocalScheduler::~LocalScheduler() {
  // Remove all jobs.
  msg("Destructing LocalScheduler");
  if ( m_pimp->m_remove_jobs ) {
    msg("Removing all jobs");
    for ( JobIdList::reverse_iterator ijid=m_pimp->m_jids.rbegin();
          ijid!=m_pimp->m_jids.rend(); ++ijid ) {
      remove(*ijid);
    }
  }
  // Delete implementation.
  delete m_pimp;
}


//**********************************************************************
// Scheduler non-const interface.
//**********************************************************************

// Add a task.

int LocalScheduler::add_task(const Application& app, const Task& tsk) {
  m_pimp->lock();
  msg("Installing task " + tsk.id().to_string());
  // Check if job has already been submitted.
  TaskJobMap& taskjobs = m_pimp->m_taskjobs;
  string xid = app.id().to_string() + " " + tsk.id().to_string();
  if ( taskjobs.find(xid) != taskjobs.end() ) {
    msg("  Job has already been submitted to build task "
        + tsk.id().to_string());
    m_pimp->unlock();
    return 1;
  }
  // Fetch the command file for building the task.
  string appdir = application_directory(app, true);
  if ( ! appdir.size() ) {
    m_pimp->
      msg("  Unable to find/create application directory:");
    msg("  " + appdir);
    m_pimp->unlock();
    return 2; 
  }
  string bldfile = appdir + dsep() + "build_task";
  Name tdir = task_directory(app, tsk);
  if ( tdir == "" ) {
    msg("  Unable to find task directory name");
    m_pimp->unlock();
    return 4;
  }
  msg("  Task directory is " + tdir);
  string taskok = tdir + "/task_installed";
  string taskfail = tdir + "/task_install_failed";
  string taskbuilding = tdir + "/task_building";
  // If the directory does not exist, then install the task.
  if ( FileStatus(tdir).exists() ) {
    msg("  Task directory already exists");
    msg("    " + tdir);
    if ( FileStatus(taskfail).exists() ) {
      msg("    Previous task installation failed--give up");
      m_pimp->unlock();
      return 5;
    } else if ( FileStatus(taskbuilding).exists() ) {
      msg("    Another scheduler is building the task--give up");
      m_pimp->unlock();
      return 5;
    } else if ( ! FileStatus(taskok).exists() ) {
      msg("    Directory contains no status files--give up");
      m_pimp->unlock();
      return 5;
    }
    msg("    Previous task installation succeeded--use it");
    m_pimp->unlock();
    return 0;
  }
  string com = "mkdir -p " + tdir;
  com += "; rm -f " + taskok + " " + taskfail;
  com += "; touch " + taskbuilding;
  SystemCommand scom(com);
  if ( scom.runout() != 0 ) {
    msg("  Error initializing status files");
    cout << scom << endl;
    return 6;
  }
  msg("  Writing task files into the task directory");
  int wstat = tsk.write_files(tdir, false);
  if ( wstat != 0 ) {
    msg("    Unable to write task files");
    m_pimp->unlock();
    return 7;
  }
  if ( ! app.has("build_task") ) {
    msg("  There is no build_task--installation complete");
    SystemCommand scom2("rm -f " + taskbuilding + "; touch " + taskok);
    if ( scom2.runout() != 0 ) {
      msg("    Error updating status files--ignore it");
      cout << scom2 << endl;
    }
    m_pimp->unlock();
    return 0;
  }
  //
  // Create job to build task.
  //
  // Create command to remove task directory.
  // Only run this if there is an error.
  SystemCommand removedir("rm -rf " + tdir);
  // Assign job ID.
  JobId jid = JobId::generate();
  if ( ! jid.is_valid() ) {
    msg("  Unable to generate job ID");
    m_pimp->unlock();
    removedir.run();
    return 11;
  }
  msg("  Job ID is " + jid.to_string());
  // Create run directory.
  Name jobdir = job_directory() + dsep() + jid.to_path();
  {
    int stat = mkfulldir(jobdir);
    if ( stat != 0 ) {
      msg("  Unable to create run directory");
      msg("    " + jobdir);
      m_pimp->unlock();
      removedir.run();
      return 12;
    }
  }
  msg("  Run directory is:");
  msg("    " + jobdir);
  // Copy the build script.
  string local_bldfile = jobdir + dsep() + "dial_build_script";
  int cstat = copy_file(bldfile, local_bldfile);
  if ( cstat ) {
    msg("Unable to copy build script " + bldfile );
    m_pimp->unlock();
    removedir.run();
    return 13;
  }
  ssystem("chmod +x " + local_bldfile);
  // Create a result file to use in case job wrapper is unable to
  // generate one.
  {
    string fresxml = jobdir + dsep() + "default_build_result.xml";
    string fresdat = "default_build_result.txt";
    if ( ! FileStatus(fresdat).exists() ) {
      Text bres(fresdat);
      bres.append("Build job was unable to generate a result");
      bres.write();
    }
    TextDataset tres("LOG", fresdat);
    tres.lock();
    assert( tres.is_valid() );
    const XmlElement* pxres = tres.xml();
    assert( pxres != 0 );
    XmlParser parser;
    parser.write(fresxml, *pxres);
    delete pxres;
  }
  // Fetch DIAL package and configuration.
  string dialpkg = Environment::current().value("DIAL_PKGNAME");
  string dialcfg = Environment::current().value("DIAL_CFGNAME");
  // Create run script. It changes to the build directory and
  // runs the build script there.
  string local_runfile = jobdir + dsep() + "dial_run_script";
  Text trun(local_runfile);
  trun.append("#!/bin/sh");
  trun.append("");
  trun.append("RUNDIR=`pwd`");
  trun.append("");
  trun.append("echo");
  trun.append("echo Change to task directory:");
  trun.append("echo " + tdir);
  trun.append("cd " + tdir);
  trun.append("pwd");
  trun.append("");
  trun.append("echo");
  trun.append("echo Run build script");
  trun.append("COM=" + jobdir + "/dial_build_script");
  trun.append("echo $COM");
  trun.append("$COM >$RUNDIR/build.log 2>&1");
  trun.append("DIAL_BUILD_STAT=$?");
  trun.append("echo Command returned $DIAL_BUILD_STAT");
  trun.append("echo Output is in build.log");
  trun.append("");
  trun.append("echo");
  trun.append("echo Updating status files");
  trun.append("");
  trun.append("if [ $DIAL_BUILD_STAT -eq 0 ]; then");
  trun.append("  touch task_installed");
  trun.append("else");
  trun.append("  touch task_install_failed");
  trun.append("fi");
  trun.append("rm -f task_building");
  trun.append("");
  trun.append("echo");
  trun.append("echo Change back to run directory:");
  trun.append("echo $RUNDIR");
  trun.append("cd $RUNDIR");
  trun.append("pwd");
  trun.append("");
  trun.append("echo");
  trun.append("echo Set up DIAL");
  trun.append("DIAL_PKGNAME=" + dialpkg);
  trun.append("DIAL_CFGNAME=" + dialcfg);
  trun.append("DIALDIR=`pkgmgr locate $DIAL_PKGNAME`");
  trun.append("echo DIALDIR = $DIALDIR");
  trun.append(". $DIALDIR/bin/dialsetup.sh $DIAL_PKGNAME $DIAL_CFGNAME");
  trun.append("");
  trun.append("echo");
  trun.append("echo Create result");
  trun.append("make_text_dataset -c LOG build.log -f result.xml");
  trun.append("DIAL_RESULT_STAT=$?");
  trun.append("echo Command returned $DIAL_RESULT_STAT");
  trun.append("if [ $DIAL_RESULT_STAT -ne 0 -a $DIAL_BUILD_STAT -eq 0 ]; then");
  trun.append("  cp default_build_result.xml result.xml");
  trun.append("fi");
  trun.append("");
  trun.append("exit $DIAL_BUILD_STAT");
  trun.write();
  ssystem("chmod +x " + local_runfile);
  // Create dummy dataset and prefs--job creator requires these.
  static TextDataset dst("JUNK", "");
  if ( ! dst.is_locked() ) dst.lock();
  static JobPreferences prf;
  if ( ! prf.is_locked() ) prf.lock();
  Job* pjob = m_pimp->m_cre.create_local_job(jid, app, tsk, dst, prf,
                                             jobdir, "build_task");
  pjob->start();
  taskjobs[xid] = jid;
  if ( pjob == 0 ) {
    msg("  Unable to create task job");
    m_pimp->unlock();
    removedir.run();
    return 14;
  }
  JobMap& jobs = m_pimp->m_jobs;
  assert( jobs.find(jid) == jobs.end() );
  jobs[jid] = pjob;
  m_pimp->m_jids.push_back(jid);
  m_pimp->unlock();
  return 0;
}


//**********************************************************************

// Remove a task job.

int LocalScheduler::
remove_task(const Application& app, const Task& tsk) {
  m_pimp->lock();
  TaskJobMap& taskjobs = m_pimp->m_taskjobs;
  string xid = app.id().to_string() + " " + tsk.id().to_string();
  TaskJobMap::const_iterator ijob = taskjobs.find(xid);
  if ( ijob == taskjobs.end() ) {
    m_pimp->unlock();
    return 1;
  }
  JobId jid = ijob->second;
  remove(jid);
  m_pimp->unlock();
  return 0;
}

//**********************************************************************

// Submit a job.

JobId LocalScheduler::submit(const Application& app,
                             const Task& tsk,
                             const Dataset& dst,
                             const JobPreferences& prf) {
  m_pimp->lock();
  msg("Job submitted:");
  if ( ! is_valid() ) {
    msg("  Scheduler is not valid");
    m_pimp->unlock();
    return JobId();
  }
  string tskdir = task_directory(app,tsk);
  if ( task_directory(app,tsk) == "" ) {
    msg("  Task directory not found");
    m_pimp->unlock();
    return JobId();
  }
  msg("  Task directory is:");
  msg("    " + tskdir);
  if ( ! dst.is_valid() ) {
    msg("  Dataset is not valid");
    m_pimp->unlock();
    return JobId();
  }
  // Fetch command line.
  string appdir = application_directory(app, true);
  if ( ! appdir.size() ) {
    msg("  Unable to find application directory");
    m_pimp->unlock();
    return JobId();
  }
  // Locate and check run script.
  string runfile = appdir + dsep() + "run";
  {
    FileStatus fstat(appdir);
    if ( ! fstat.is_directory() || !fstat.is_readable() ) {
      msg("  Unable to open application directory:");
      msg("    " + appdir);
      m_pimp->unlock();
      return JobId();
    }
  }
  {
    FileStatus fstat(runfile);
    if ( ! fstat.is_executable() ) {
      msg("  Unable to locate executable run script");
      msg("    " + runfile);
      m_pimp->unlock();
      return JobId();
    }
  }
  msg("  Run script is:");
  msg("    " + runfile);
  // Assign job ID.
  JobId jid = JobId::generate();
  if ( ! jid.is_valid() ) {
    msg("  Unable to generate job ID");
    m_pimp->unlock();
    return jid;
  }
  msg("  Job ID is " + jid.to_string());
  // Create run directory.
  Name jobdir = job_directory() + dsep() + jid.to_path();
  {
    int stat = mkfulldir(jobdir);
    if ( stat != 0 ) {
      msg("  Unable to create run directory");
      msg("    " + jobdir);
      m_pimp->unlock();
      return JobId();
    }
  }
  msg("  Run directory is:");
  msg("    " + jobdir);
  // Copy the run script.
  string local_runfile = jobdir + dsep() + "dial_run_script";
  copy_file(runfile, local_runfile);
  // Write the task directory into the job directory.
  {
    Text tskloc(jobdir + dsep() + "taskdir");
    tskloc.append(tskdir);
    tskloc.write();
  }
  // Write the job ID into the job directory.
  {
    Text jidloc(jobdir + dsep() + "jid");
    jidloc.append(jid.to_string());
    jidloc.write();
  }
  // Create job.
  JobMap& jobs = m_pimp->m_jobs;
  assert( jobs.find(jid) == jobs.end() );
  Job* pjob =
    m_pimp->m_cre.create_local_job(jid, app, tsk, dst, prf,
                                   jobdir, "run");
  if ( pjob == 0 ) {
    msg("  Unable to create job");
    m_pimp->unlock();
    return JobId();
  }
  jobs[jid] = pjob;
  m_pimp->m_jids.push_back(jid);
  // Add job to repository.
  JobRepository* psjr = job_repository();
  if ( psjr!=0 && psjr->is_valid() ) {
    msg("  Updating repository with job " + jid.to_string() );
    JobId chkid = psjr->insert(pjob, false);
    if ( ! chkid.is_valid() ) {
      msg("    Insert failed: " + psjr->error_message());
    }
    assert( chkid == jid );
  }
  m_pimp->unlock();
  return jid;
}

//**********************************************************************

// Kill a job.

int LocalScheduler::kill(JobId jid) {
  m_pimp->lock();
  int stat = job(jid).kill();
  m_pimp->unlock();
  return stat;
}

//**********************************************************************

// Remove a job.

int LocalScheduler::remove(JobId jid) {
  m_pimp->lock();
  msg("Removing job " + jid.to_string() );
  Job& xjob = job(jid);
  if ( ! xjob.is_valid() ) {
    msg("  Job is invalid");
    m_pimp->unlock();
    return 1;
  }
  if ( xjob.is_running() ) {
    xjob.update();
    if ( xjob.is_running() ) {
      msg("  Kill running job");
      xjob.kill();
      xjob.update();
      // Maybe we should wait and try again....
      if ( xjob.is_running() ) {
        msg("  Job kill failed");
        m_pimp->unlock();
	return 2;
      }
    }
  }
  // Drop job from lists.
  JobMap::iterator ijob = m_pimp->m_jobs.find(jid);
  if ( ijob == m_pimp->m_jobs.end() ) {
    msg("  Unable to locate job in job map");
    m_pimp->unlock();
    return 3;
  }
  JobIdList::iterator ijid =
    find(m_pimp->m_jids.begin(), m_pimp->m_jids.end(), jid);
  if ( ijid == m_pimp->m_jids.end() ) {
    msg("  Unable to locate job ID in job ID list");
    m_pimp->unlock();
    return 4;
  }
  m_pimp->m_jobs.erase(ijob);
  m_pimp->m_jids.erase(ijid);
  // Delete job.
  delete &xjob;
  // Delete run directory.
  Name jobdir = job_directory() + dsep() + jid.to_path();
  string com = "rm -rf " + jobdir;
  system(com.c_str());
  if ( FileStatus(jobdir).exists() ) {
    msg("  Unable to delete run directory:");
    msg("    " + jobdir );
    m_pimp->unlock();
    return 5;
  }
  // Remove job from repository.
  if ( job_repository()->is_valid() ) {
    if ( job_repository()->remove(jid) != 0 ) {
      msg("  Removal from JR failed.");
      msg("    " + job_repository()->error_message() );
    }
  }
  m_pimp->unlock();
  return 0;
}

//**********************************************************************
// Scheduler const interface
//**********************************************************************

// Is an application available?

bool LocalScheduler::has_application(const Application& app) const {
  m_pimp->lock();
  Name dir = application_directory(app, true);
  m_pimp->unlock();
  return dir != "";
}

//**********************************************************************

// Is a task installed?

bool LocalScheduler::
has_task(const Application& app, const Task& tsk) const {
  m_pimp->lock();
  Name dir = task_directory(app, tsk);
  Name file = dir + dsep() + "task_installed";
  bool stat = check_file(file);
  m_pimp->unlock();
  return stat;
}

//**********************************************************************

// Return the ID of the job used to build the task.

JobId LocalScheduler::
task_job(const Application& app, const Task& tsk) const {
  m_pimp->lock();
  TaskJobMap& taskjobs = m_pimp->m_taskjobs;
  string xid = app.id().to_string() + " " + tsk.id().to_string();
  TaskJobMap::const_iterator ijob = taskjobs.find(xid);
  if ( ijob == taskjobs.end() ) {
    m_pimp->unlock();
    return JobId();
  }
  JobId jid = ijob->second;
  m_pimp->unlock();
  return jid;
}
  
//**********************************************************************

// Return the list of jobs.

JobIdList LocalScheduler::jobs() const {
  m_pimp->lock();
  JobIdList jids = m_pimp->m_jids;
  m_pimp->unlock();
  return jids;
}

//**********************************************************************

// Return a job.

Job& LocalScheduler::job(JobId jid) const {
  m_pimp->lock();
  static Job BADJOB;
  JobMap::const_iterator ijob = m_pimp->m_jobs.find(jid);
  if ( ijob == m_pimp->m_jobs.end() ) {
    m_pimp->unlock();
    return BADJOB;
  }
  Job& job = *ijob->second;
  m_pimp->unlock();
  return job;
}

//**********************************************************************

string LocalScheduler::job_repository_connection() const {
  return m_pimp->jr_conn;
}

//**********************************************************************

JobRepository* LocalScheduler::job_repository() const {
  return m_pimp->pjr;
}

//**********************************************************************

// XML.

const XmlElement* LocalScheduler::xml() const {
  m_pimp->lock();
  auto_ptr<XmlElement> pele(new XmlElement(xml_name()));
  pele->add_attribute_as_bool("remove_jobs", m_pimp->m_remove_jobs);
  pele->add_attribute("jobrep", job_repository_connection());
  const XmlElement* pxjcr = job_creator().xml();
  if ( pxjcr == 0 ) {
    m_pimp->unlock();
    return 0;
  }
  pele->add_child(pxjcr);
  m_pimp->unlock();
  return pele.release();
}

//**********************************************************************
// Extension of interface
//**********************************************************************

// Validity.

bool LocalScheduler::is_valid() const {
  m_pimp->lock();
  bool stat = m_pimp!=0 && m_pimp->m_valid;
  m_pimp->unlock();
  return stat;
}

//**********************************************************************

// Log a message.

void LocalScheduler::msg(string txt) const {
  m_pimp->msg(txt);
}

//**********************************************************************

// Job creator.

const JobCreator& LocalScheduler::job_creator() const {
  return m_pimp->m_cre;
}
//**********************************************************************

// Return the directory where an application may be found.

Name LocalScheduler::
application_directory(const Application& app, bool create) const {
  if ( ! is_valid() ) {
    msg("  Invalid scheduler");
    return "";
  }
  if ( ! app.is_valid() ) {
    msg("  Invalid application");
    return "";
  }
  Name aid = app.id().to_string();
  if ( aid == "" ) {
    msg("  Invalid application ID");
    return "";
  }
  Name dir = m_pimp->m_appdir + dsep() + aid;
  if ( ! FileStatus(dir).exists() && create ) {
    int wstat = app.write_files(dir);
    if ( wstat ) {
      msg("  Unable to write application files");
      msg("    " + app.error_message());
      return "";
    }
  }
  if ( ! check_dir(dir) ) {
    msg("  Application directory check failed");
    msg("    " + dir);
    return "";
  }
  return dir;
}

//**********************************************************************

// Return the directory where a task is installed.

Name LocalScheduler::
task_directory(const Application& app, const Task& tsk) const {
  m_pimp->lock();
  if ( ! is_valid() ) {
    msg("  Scheduler is not valid");
    m_pimp->unlock();
    return "";
  }
  if ( ! tsk.is_valid() ) {
    msg("  Task is not valid");
    m_pimp->unlock();
    return "";
  }
  if ( ! tsk.id().is_global() ) {
    msg("  Task ID is not global");
    m_pimp->unlock();
    return "";
  }
  Name tdir = m_pimp->m_tskdir + dsep() + 
              app.id().to_string() + dsep() +
              tsk.id().to_string();
  m_pimp->unlock();
  return tdir;
}

//**********************************************************************

// Return the directory where job directories are found.

Name LocalScheduler::job_directory() const {
  return m_pimp->m_jobdir;
}

//**********************************************************************

Text LocalScheduler::log() const {
  m_pimp->lock();
  Text tmplog = m_pimp->m_log;
  m_pimp->unlock();
  return tmplog;
}

//**********************************************************************

// Output stream.

ostream& LocalScheduler::ostr(ostream& lhs) const {
  m_pimp->lock();
  lhs << "LocalScheduler" << endl;
  ostream& rstr = job_report(lhs);
  m_pimp->unlock();
  return rstr;
}

//**********************************************************************