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

#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;
using HeuristicLab.Hive.JobBase;

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, true);
    }

    /// <summary>
    /// Adds a new job into the database
    /// </summary>
    /// <param name="job"></param>
    /// <param name="overrideUserid">if false, job.JobInfo.UserId will not be overridden</param>
    /// <returns></returns>
    public ResponseObject<JobDto> AddNewJob(SerializedJob job, bool overrideUserid) {
      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;
        if (overrideUserid) {
          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;
    }

    public ResponseObject<JobDto> AddNewJob(SerializedJob job) {
      return AddNewJob(job, true);
    }

    /// <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;
    //}

    public ResponseObject<SerializedJob> GetLastSerializedResult(Guid jobId) {
      ResponseObject<SerializedJob> response = new ResponseObject<SerializedJob>();
      JobDto job = DaoLocator.JobDao.FindById(jobId);

      response.Obj = new SerializedJob();
      response.Obj.JobInfo = job;

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

      return response;
    }

    public ResponseObject<SerializedJob> GetSnapshotResult(Guid jobId) {
      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 (job.State == JobState.SnapshotRequested || job.State == JobState.SnapshotSent) {
        response.StatusMessage = ResponseStatus.GetSnapshotResult_JobResultNotYetThere;
        return response;
      }

      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() {
          Id = 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);
    }

    public ResponseObject<JobResultList> GetChildJobResults(Guid? parentJobId, bool recursive, bool includeParent) {
      ResponseObject<JobResultList> response = new ResponseObject<JobResultList>();
      JobResultList jobResultList = new JobResultList();

      IList<JobDto> jobs = DaoLocator.JobDao.FindJobsByParentId(parentJobId, recursive).ToList();
      if (!parentJobId.HasValue) {
        jobs = jobs.Where(job => job.UserId == ServiceLocator.GetAuthorizationManager().UserId).ToList();
      }

      if (includeParent && parentJobId.HasValue) {
        jobs.Add(DaoLocator.JobDao.FindById(parentJobId.Value));
      }

      foreach (JobDto job in jobs) {
        if (job != null) {
          jobResultList.Add(new JobResult() {
            Id = job.Id,
            State = job.State,
            DateCreated = job.DateCreated,
            DateCalculated = job.DateCalculated,
            DateFinished = job.DateFinished,
            Exception = job.Exception,
            Percentage = job.Percentage,
            ParentJobId = job.ParentJob != null ? new Nullable<Guid>(job.ParentJob.Id) : null
          });
        }
      }
      
      response.Obj = jobResultList;
      return response;
    }

    public ResponseObject<JobDto> AddChildJob(Guid parentJobId, SerializedJob serializedJob) {
      JobDto parentJob = DaoLocator.JobDao.FindById(parentJobId);

      serializedJob.JobInfo.ParentJob = parentJob;
      serializedJob.JobInfo.UserId = parentJob.UserId;
      serializedJob.JobInfo.AssignedResourceIds = parentJob.AssignedResourceIds;

      return AddNewJob(serializedJob, false);
    }

    public ResponseObject<JobDto> PauseJob(SerializedJob serializedJob) {
      JobDto jobDto = DaoLocator.JobDao.FindById(serializedJob.JobInfo.Id);

      jobDto.State = serializedJob.JobInfo.State;
      jobDto.Percentage = serializedJob.JobInfo.Percentage;
      

      DaoLocator.JobDao.SetBinaryJobFile(serializedJob.JobInfo.Id, serializedJob.SerializedJobData);
      DaoLocator.JobDao.Update(jobDto);

      DaoLocator.JobDao.UnAssignSlaveToJob(jobDto.Id);

      return new ResponseObject<JobDto>() {
        Obj = jobDto,
        StatusMessage = ResponseStatus.Ok
      };
    }

    #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


    

  }
}