source: branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/JobManager.cs @ 4333

#region License Information

/* HeuristicLab
 * Copyright (C) 2002-2008 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */

#endregion

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using HeuristicLab.Hive.Contracts.Interfaces;
using HeuristicLab.Hive.Contracts.BusinessObjects;
using HeuristicLab.Hive.Contracts;
using HeuristicLab.Hive.Server.DataAccess;
using HeuristicLab.Hive.Server.Core.InternalInterfaces;
using HeuristicLab.DataAccess.Interfaces;
using System.Data;
using System.IO;
using HeuristicLab.Tracing;
using System.Transactions;
using HeuristicLab.Hive.Server.LINQDataAccess;
using IsolationLevel = System.Transactions.IsolationLevel;
using HeuristicLab.Hive.Contracts.ResponseObjects;

namespace HeuristicLab.Hive.Server.Core {
  internal class JobManager : IJobManager, IInternalJobManager {
    private ILifecycleManager lifecycleManager;

    #region IJobManager Members

    public JobManager() {
      lifecycleManager = ServiceLocator.GetLifecycleManager();

      lifecycleManager.RegisterStartup(new EventHandler(lifecycleManager_OnStartup));
      lifecycleManager.RegisterShutdown(new EventHandler(lifecycleManager_OnShutdown));
    }

    private void CheckForDeadJobs() {
      Logger.Info("Searching for dead Jobs");

      List<JobDto> allJobs = new List<JobDto>(DaoLocator.JobDao.FindAll());
      foreach (JobDto curJob in allJobs) {
        if (curJob.State != JobState.Calculating && curJob.State != JobState.Finished) {
          DaoLocator.JobDao.SetJobOffline(curJob);
        }
      }
    }

    private void lifecycleManager_OnStartup(object sender, EventArgs e) {
      Logger.Info("Startup Event Fired, Checking DB for consistency");
      CheckForDeadJobs();
      Logger.Info("Startup Event Done");
    }

    private void lifecycleManager_OnShutdown(object sender, EventArgs e) {
      Logger.Info("Startup Event Fired, Checking DB for consistency");
      CheckForDeadJobs();
      Logger.Info("Startup Event Done");
    }

    /// <summary>
    /// returns all jobs stored in the database
    /// </summary>
    /// <returns></returns>
    public ResponseList<JobDto> GetAllJobs() {
      ResponseList<JobDto> response = new ResponseList<JobDto>();
      response.List = new List<JobDto>(DaoLocator.JobDao.FindAll());
      return response;
    }

    public ResponseList<JobDto> GetAllJobsWithFilter(JobState jobState, int offset, int count) {
      ResponseList<JobDto> response = new ResponseList<JobDto>();
      response.List = new List<JobDto>(DaoLocator.JobDao.FindWithLimitations(jobState, offset, count));
      return response;
    }

    /// <summary>
    /// Gets the streamed job
    /// </summary>
    /// <param name="jobId"></param>
    /// <returns></returns>
    public Stream GetJobStreamById(Guid jobId) {
      return DaoLocator.JobDao.GetSerializedJobStream(jobId);
    }

    /// <summary>
    /// returns the job with the specified id
    /// </summary>
    /// <returns></returns>
    public ResponseObject<JobDto> GetJobById(Guid jobId) {
      ResponseObject<JobDto> response = new ResponseObject<JobDto>();

      response.Obj = DaoLocator.JobDao.FindById(jobId);
      if (response.Obj != null) {
        response.StatusMessage = ResponseStatus.Ok;
      } else {
        response.StatusMessage = ResponseStatus.GetJobById_JobDoesNotExist;
      }

      return response;
    }

    /// <summary>
    /// Returns JobDto object with Slave-Object attached
    /// </summary>
    public ResponseObject<JobDto> GetJobByIdWithDetails(Guid jobId) {
      ResponseObject<JobDto> job = new ResponseObject<JobDto>();
      job.Obj = DaoLocator.JobDao.FindById(jobId);
      if (job.Obj != null) {
        job.StatusMessage = ResponseStatus.Ok;

        job.Obj.Slave = DaoLocator.SlaveDao.GetSlaveForJob(jobId);
      } else {
        job.StatusMessage = ResponseStatus.GetJobByIdWithDetails_JobDoesNotExist;
      }
      return job;
    }

    public ResponseObject<JobDto> AddJobWithGroupStrings(SerializedJob job, IEnumerable<string> resources) {
      ISlaveGroupDao cgd = DaoLocator.SlaveGroupDao;
      foreach (string res in resources) {
        foreach (SlaveGroupDto cg in cgd.FindByName(res)) {
          job.JobInfo.AssignedResourceIds.Add(cg.Id);
        }
      }
      return AddNewJob(job);
    }

    /// <summary>
    /// Adds a new job into the database
    /// </summary>
    /// <param name="job"></param>
    /// <returns></returns>
    public ResponseObject<JobDto> AddNewJob(SerializedJob job) {
      ResponseObject<JobDto> response = new ResponseObject<JobDto>();

      if (job != null && job.JobInfo != null) {
        if (job.JobInfo.State != JobState.Offline) {
          response.StatusMessage = ResponseStatus.AddNewJob_JobStateMustBeOffline;
          return response;
        }
        if (job.JobInfo.Id != Guid.Empty) {
          response.StatusMessage = ResponseStatus.AddNewJob_JobIdMustNotBeSet;
          return response;
        }
        if (job.SerializedJobData == null) {
          response.StatusMessage = ResponseStatus.AddNewJob_JobNull;
          return response;
        }

        job.JobInfo.DateCreated = DateTime.Now;
        job.JobInfo.UserId = ServiceLocator.GetAuthorizationManager().UserId;
        DaoLocator.JobDao.InsertWithAttachedJob(job);
        DaoLocator.PluginInfoDao.InsertPluginDependenciesForJob(job.JobInfo);

        response.Obj = job.JobInfo;
      } else {
        response.StatusMessage = ResponseStatus.AddNewJob_JobNull;
      }

      return response;
    }

    /// <summary>
    /// Removes a job from the database
    /// 
    /// [chn] this is currently not used anywhere -> check redundancy with AbortJob
    /// </summary>
    /// <param name="jobId"></param>
    /// <returns></returns>
    //public Response RemoveJob(Guid jobId) {
    //  Response response = new Response();

    //  JobDto job = DaoLocator.JobDao.FindById(jobId);
    //  if (job == null) {
    //    response.StatusMessage = ResponseStatusMessage.RemoveJob_JobDoesNotExist;
    //    return response;
    //  }
    //  DaoLocator.JobDao.Delete(job);

    //  return response;
    //}

    ///
    /// [chn] currently not used anywhere
    /// 
    //public ResponseObject<JobDto> GetLastJobResultOf(Guid jobId) {
    //  ResponseObject<JobDto> result = new ResponseObject<JobDto>();

    //  result.Obj = DaoLocator.JobDao.FindById(jobId);
    //  if (result.Obj == null) {
    //    result.StatusMessage = ResponseStatusMessage.
    //  }

    //  return result;
    //}

    // [chn] [refactor] why does this method handle 3 different cases? wouldn't 3 methods be easier?
    //Requested means: there MUST be a job result which gets sent back
    // requested==true: resultat mit job
    // requested==false: 
    public ResponseObject<SerializedJob> GetLastSerializedResult(Guid jobId, bool requested, bool snapshot) {
      ResponseObject<SerializedJob> response = new ResponseObject<SerializedJob>();

      JobDto job = DaoLocator.JobDao.FindById(jobId);

      //if it's a snapshot but the result hasn't reached the server yet...
      if (snapshot && (job.State == JobState.SnapshotRequested || job.State == JobState.SnapshotSent)) {
        response.StatusMessage = ResponseStatus.GetLastSerializedResult_JobResultNotYetThere;

        return response;
      }

      //if it's NOT a snapshot, NEITHER request NOR is it finished
      if (!requested && !snapshot && job.State != JobState.Finished) {
        response.StatusMessage = ResponseStatus.GetLastSerializedResult_JobResultNotYetThere;

        return response;
      }

      //every other case - snapshot, job finished or it's requested
      response.Obj = new SerializedJob();
      response.Obj.JobInfo = job;

      response.Obj.SerializedJobData = DaoLocator.JobDao.GetBinaryJobFile(jobId);

      return response;
    }

    public Response RequestSnapshot(Guid jobId) {
      Response response = new Response();

      JobDto job = DaoLocator.JobDao.FindById(jobId);
      if (job.State == JobState.SnapshotRequested || job.State == JobState.SnapshotSent) {
        response.StatusMessage = ResponseStatus.RequestSnapshot_SnapshotAlreadyRequested;
        return response;
      }
      if (job.State != JobState.Calculating) {
        response.StatusMessage = ResponseStatus.RequestSnapshot_JobIsNotBeeingCalculated;
        return response;
      }
      // job is in correct state
      job.State = JobState.SnapshotRequested;
      DaoLocator.JobDao.Update(job);

      return response;
    }

    /// <summary>
    /// [chn] rename to RequestAbortJob?
    /// </summary>
    public Response AbortJob(Guid jobId) {
      Response response = new Response();

      JobDto job = DaoLocator.JobDao.FindById(jobId);
      if (job == null) {
        //response.Success = false;
        response.StatusMessage = ResponseStatus.AbortJob_JobDoesNotExist;
        return response; // no commit needed
      }
      if (job.State == JobState.Aborted) {
        //response.Success = true;
        response.StatusMessage = ResponseStatus.AbortJob_AbortAlreadyRequested;
        return response; // no commit needed
      }
      if (job.State != JobState.Calculating && job.State != JobState.SnapshotRequested && job.State != JobState.SnapshotSent) {
        //response.Success = false;
        response.StatusMessage = ResponseStatus.AbortJob_JobIsNotBeeingCalculated;
        return response; // no commit needed
      }
      // job is in correct state
      job.State = JobState.Aborted;
      DaoLocator.JobDao.Update(job);

      return response;
    }

    /// <summary>
    /// Returns the current state for all jobs requested
    /// </summary>
    /// <param name="jobIds"></param>
    /// <returns></returns>
    public ResponseObject<JobResultList> GetJobResults(IEnumerable<Guid> jobIds) {
      ResponseObject<JobResultList> response = new ResponseObject<JobResultList>();
      JobResultList jobResultList = new JobResultList();
      IEnumerable<JobDto> jobs = DaoLocator.JobDao.FindJobsById(jobIds);
      foreach (JobDto job in jobs) {
        jobResultList.Add(new JobResult() {
          JobId = job.Id,
          State = job.State,
          DateCalculated = job.DateCalculated,
          DateFinished = job.DateFinished,
          Exception = job.Exception,
          Percentage = job.Percentage
        });
      }
      response.Obj = jobResultList;
      return response;
    }

    public byte[] GetSerializedJobDataById(Guid jobId) {
      return DaoLocator.JobDao.GetBinaryJobFile(jobId);
    }

    public void SetSerializedJobDataById(Guid jobId, byte[] data) {
      DaoLocator.JobDao.SetBinaryJobFile(jobId, data);
    }

    #endregion

    #region Project handling (currently unimplemented)

    public ResponseList<ProjectDto> GetAllProjects() {
      throw new NotImplementedException();
    }

    private Response CreateUpdateProject(ProjectDto project) {
      throw new NotImplementedException();
    }

    public Response CreateProject(ProjectDto project) {
      return CreateUpdateProject(project);
    }

    public Response ChangeProject(ProjectDto project) {
      return CreateUpdateProject(project);
    }

    public Response DeleteProject(Guid projectId) {
      throw new NotImplementedException();
    }

    public ResponseList<JobDto> GetJobsByProject(Guid projectId) {
      throw new NotImplementedException();
    }
    #endregion
  }
}