source: branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/SlaveCommunicator.cs @ 5213

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2010 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.IO;
using System.Linq;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using System.Transactions;
using HeuristicLab.Hive.Contracts;
using HeuristicLab.Hive.Contracts.BusinessObjects;
using HeuristicLab.Hive.Contracts.Interfaces;
using HeuristicLab.Hive.Contracts.ResponseObjects;
using HeuristicLab.Hive.Server.Core.InternalInterfaces;
using HeuristicLab.PluginInfrastructure;
using System.Diagnostics;
using HeuristicLab.Hive.Tracing;
using HeuristicLab.PluginInfrastructure.Manager;
using System.Web;
using System.Web.SessionState;

namespace HeuristicLab.Hive.Server.Core {
  /// <summary>
  /// The SlaveCommunicator manages the whole communication with the slave
  /// </summary>
  public class SlaveCommunicator : ISlaveCommunicator, IInternalSlaveCommunicator {
    private ILifecycleManager lifecycleManager;
    private IInternalJobManager jobManager;
    private IScheduler scheduler;

    private static object slaveLocker = new object();

    /// <summary>
    /// Initialization of the Adapters to the database
    /// Initialization of Eventhandler for the lifecycle management
    /// Initialization of lastHearbeats Dictionary
    /// </summary>
    public SlaveCommunicator() {
      Logger.Debug("ServiceCommunicator instantiated");
      lifecycleManager = ServiceLocator.GetLifecycleManager();
      jobManager = ServiceLocator.GetJobManager() as IInternalJobManager;
      scheduler = ServiceLocator.GetScheduler();

      lifecycleManager.ServerHeartbeat += new EventHandler(lifecycleManager_OnServerHeartbeat);
    }

    /// <summary>
    /// Check if online slaves send their hearbeats
    /// if not -> set them offline and check if they where calculating a job
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="e"></param>
    public void lifecycleManager_OnServerHeartbeat(object sender, EventArgs e) {
      // this block can conflict with the heartbeats from a slave (which also updates the slave-records)
      lock (slaveLocker) {
        Logger.Debug("Server Heartbeat ticked");
        List<Guid> slaveIds = DaoLocator.SlaveDao.FindAll().Select(s => s.Id).ToList();
        foreach (Guid slaveId in slaveIds) {
          using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = ApplicationConstants.ISOLATION_LEVEL_SCOPE })) {
            SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);

            TimeSpan diff = TimeSpan.MaxValue;
            if(slave.LastHeartbeat != null )
              diff = DateTime.Now.Subtract(slave.LastHeartbeat.Value);

            if (slave.State != SlaveState.Offline && diff.TotalSeconds > ApplicationConstants.HEARTBEAT_MAX_DIFF_SECONDS) {
              // if slave calculated jobs, the job must be reset
              Logger.Info("Slave timed out and is on RESET (no message for " + diff.TotalSeconds + " seconds.)");
              foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfSlave(slave)) {
                DaoLocator.JobDao.SetJobOffline(job);
              }
              Logger.Debug("setting slave offline");
              slave.State = SlaveState.Offline;

              DaoLocator.SlaveDao.Update(slave);

              Logger.Debug("removing it from the heartbeats list");
            }

            scope.Complete();
          } // using
        } // foreach

        using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = ApplicationConstants.ISOLATION_LEVEL_SCOPE })) {
          var timeoutJobs = DaoLocator.JobDao.FindTimeoutJobs();
          foreach (var job in timeoutJobs) {
            DaoLocator.JobDao.SetJobOffline(job);
          }
          scope.Complete();
        }

        Logger.Debug("Server Heartbeat ended");
      } // lock
    }

    #region ISlaveCommunicator Members

    /// <summary>
    /// Login process for the slave
    /// A hearbeat entry is created as well (login is the first hearbeat)
    /// </summary>
    /// <param name="slaveInfo"></param>
    /// <returns></returns>
    public Response Login(SlaveDto slave) {
      Response response = new Response();

      SlaveDto dbSlave = DaoLocator.SlaveDao.FindById(slave.Id);

      slave.CalendarSyncStatus = dbSlave != null ? dbSlave.CalendarSyncStatus : CalendarState.NotAllowedToFetch;
      slave.State = SlaveState.Idle;
      slave.Login = DateTime.Now;

      if (dbSlave == null) {
        DaoLocator.SlaveDao.Insert(slave);
      } else {
        DaoLocator.SlaveDao.Update(slave);
      }

      return response;
    }

    public ResponseCalendar GetCalendar(Guid slaveId) {
      ResponseCalendar response = new ResponseCalendar();

      SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);
      if (slave == null) {
        response.StatusMessage = ResponseStatus.GetCalendar_ResourceDoesNotExist;
        return response;
      }

      response.ForceFetch = (slave.CalendarSyncStatus == CalendarState.ForceFetch);

      IEnumerable<AppointmentDto> appointments = DaoLocator.UptimeCalendarDao.GetCalendarForSlave(slave);
      if (appointments.Count() == 0) {
        response.StatusMessage = ResponseStatus.GetCalendar_NoCalendarFound;
      } else {
        response.Appointments = appointments;
      }

      slave.CalendarSyncStatus = CalendarState.Fetched;
      DaoLocator.SlaveDao.Update(slave);
      return response;
    }

    public Response SetCalendarStatus(Guid slaveId, CalendarState state) {
      Response response = new Response();
      SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);
      if (slave == null) {
        response.StatusMessage = ResponseStatus.GetCalendar_ResourceDoesNotExist;
        return response;
      }

      slave.CalendarSyncStatus = state;
      DaoLocator.SlaveDao.Update(slave);

      return response;
    }

    /// <summary>
    /// The slave has to send regulary heartbeats
    /// this hearbeats will be stored in the heartbeats dictionary
    /// check if there is work for the slave and send the slave a response if he should pull a job
    /// </summary>
    /// <param name="heartbeatData"></param>
    /// <returns></returns>
    public ResponseHeartBeat ProcessHeartBeat(HeartBeatData heartbeatData) {
      lock (slaveLocker) {
        Logger.Debug("BEGIN Processing Heartbeat for Slave " + heartbeatData.SlaveId);
        ResponseHeartBeat response = new ResponseHeartBeat();
        response.ActionRequest = new List<MessageContainer>();

        using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = ApplicationConstants.ISOLATION_LEVEL_SCOPE })) {
          SlaveDto slave = UpdateSlaveData(heartbeatData);
          lock (DefaultScheduler.locker) {
            DaoLocator.SlaveDao.Update(slave);
          }

          //ProcessJobProgress(heartbeatData, response);
          response.ActionRequest = ProcessJobProgress(heartbeatData);

          //check if new Cal must be loaded
          if (slave.CalendarSyncStatus == CalendarState.Fetch || slave.CalendarSyncStatus == CalendarState.ForceFetch) {
            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchOrForceFetchCalendar));
            Logger.Info("fetch or forcefetch sent");
          }

          // check if slave has a free core for a new job
          // if true, ask scheduler for a new job for this slave
          Logger.Debug(" BEGIN Looking for Slave Jobs");
          if (this.IsAllowedToSendJobs() &&
              slave.IsAllowedToCalculate &&
              heartbeatData.FreeCores > 0 &&
              scheduler.ExistsJobForSlave(heartbeatData)) {
            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
          } else {
            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
          }

          scope.Complete();
        }
        Logger.Debug(" END Processed Heartbeat for Slave " + heartbeatData.SlaveId);
        return response;
      }
    }

    /// <summary>
    /// Returns true if there are enough resources to send a job
    /// There should not be too many jobs sent simultaniously
    /// </summary>
    private bool IsAllowedToSendJobs() {
      return true; // try out without limit
      //return lifecycleManager.JobsCurrentlyTransferring < ApplicationConstants.MAX_JOB_TRANSFER_COUNT;
    }

    private SlaveDto UpdateSlaveData(HeartBeatData heartbeatData) {
      SlaveDto slave = DaoLocator.SlaveDao.FindById(heartbeatData.SlaveId);
      if (slave == null) {
        slave = new SlaveDto() { Id = heartbeatData.SlaveId };
        Login(slave);
        slave = DaoLocator.SlaveDao.FindById(heartbeatData.SlaveId);
      }

      slave.NrOfFreeCores = heartbeatData.FreeCores;
      slave.FreeMemory = heartbeatData.FreeMemory;
      slave.IsAllowedToCalculate = heartbeatData.IsAllowedToCalculate;
      slave.LastHeartbeat = DateTime.Now;

      slave.State = heartbeatData.JobProgress.Count > 0 ? SlaveState.Calculating : SlaveState.Idle;
      return slave;
    }

    private List<MessageContainer> ProcessJobProgress(HeartBeatData heartbeatData) {
      List<MessageContainer> actions = new List<MessageContainer>();

      // find all the jobs in jobProgress which are not in the database -> they are not supposed to be calculated by this slave
      //IEnumerable<Guid> jobsToAbort = GetJobsNotCalculatedByThisSlave(heartbeatData.SlaveId, heartbeatData.JobProgress.Keys);
      //foreach (Guid jobId in jobsToAbort) {
      //  Logger.Error("Job shall not be caculated by this slave or does not exist in DB: " + jobId);
      //  actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobId));
      //  heartbeatData.JobProgress.Remove(jobId);
      //}

      // process all the remaining jobProgresses
      foreach (var jobProgress in heartbeatData.JobProgress) {
        JobDto curJob = DaoLocator.JobDao.FindById(jobProgress.Key);
        if (curJob == null) {
          // job does not exist in db
          Logger.Error("Job does not exist in DB: " + jobProgress.Key);
          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
        } else {
          curJob.Slave = DaoLocator.SlaveDao.GetSlaveForJob(curJob.Id);
          Guid id = curJob.Slave != null ? curJob.Slave.Id : Guid.Empty;
          if (curJob.Slave == null || curJob.Slave.Id != heartbeatData.SlaveId) {
            // assigned slave does not match heartbeat
            Logger.Error("The slave " + heartbeatData.SlaveId + " is not supposed to calculate Job: " + curJob.ToString());
            actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
          } else {
            // save job execution time
            curJob.ExecutionTime = jobProgress.Value;
            curJob.LastHeartbeat = DateTime.Now;

            if (curJob.State == JobState.Aborted) {
              // a request to abort the job has been set
              Logger.Error("Job is in state aborted, send AbortJob: " + curJob.Id);
              actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
              DaoLocator.JobDao.UnAssignSlaveToJob(curJob.Id);
            } else if (curJob.State == JobState.SnapshotRequested) {
              // a request for a snapshot has been set
              actions.Add(new MessageContainer(MessageContainer.MessageType.RequestSnapshot, curJob.Id));
              curJob.State = JobState.SnapshotSent;
            }
          }
          DaoLocator.JobDao.Update(curJob);
        }
      }
      return actions;
    }

    /// <summary>
    /// Returns the jobIds of the jobs which are not assigned to this slave in the database (they either are not stored in DB or they are assigned
    /// </summary>
    private IEnumerable<Guid> GetJobsNotCalculatedByThisSlave(Guid slaveId, IEnumerable<Guid> jobIds) {
      IEnumerable<Guid> activeJobsOfSlave = DaoLocator.JobDao.FindActiveJobsOfSlave(DaoLocator.SlaveDao.FindById(slaveId)).Select(j => j.Id);
      return jobIds.Except(activeJobsOfSlave).ToList();
    }

    /// <summary>
    /// Process the Job progress sent by a slave
    /// [chn] this method needs to be refactored, because its a performance hog
    /// 
    /// what it does:
    /// (1) find out if the jobs that should be calculated by this slave (from db) and compare if they are consistent with what the joblist the slave sent
    /// (2) find out if every job from the joblist really should be calculated by this slave
    /// (3) checks if a job should be aborted and issues Message
    /// (4) update job-progress and write to db
    /// (5) if snapshot is requested, issue Message
    /// 
    /// (6) for each job from DB, check if there is a job from slave (again).
    /// (7) if job matches, it is removed from newAssigneJobs
    /// (8) if job !matches, job's TTL is reduced by 1, 
    /// (9) if TTL==0, job is set to Abort (save to DB), and Message to Abort job is issued to slave
    /// 
    /// 
    /// 
    /// quirks:
    /// (1) the response-object is modified during the foreach-loop (only last element counts)
    /// (2) state Abort results in Finished. This should be: AbortRequested, Aborted.
    /// </summary>
    /// <param name="heartbeatData"></param>
    /// <param name="response"></param>
    //private void ProcessJobProgress(HeartBeatData heartbeatData, ResponseHeartBeat response) {
    //  Logger.Debug("Started for Slave " + heartbeatData.SlaveId);
    //  List<JobDto> jobsOfSlave = new List<JobDto>(DaoLocator.JobDao.FindActiveJobsOfSlave(DaoLocator.SlaveDao.FindById(heartbeatData.SlaveId)));
    //  if (heartbeatData.JobProgress != null && heartbeatData.JobProgress.Count > 0) {
    //    if (jobsOfSlave == null || jobsOfSlave.Count == 0) {
    //      foreach (Guid jobId in heartbeatData.JobProgress.Keys) {
    //        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobId));
    //      }

    //      Logger.Error("There is no job calculated by this user " + heartbeatData.SlaveId + ", advise him to abort all");
    //      return;
    //    }

    //    foreach (KeyValuePair<Guid, TimeSpan> jobProgress in heartbeatData.JobProgress) {
    //      JobDto curJob = DaoLocator.JobDao.FindById(jobProgress.Key);
    //      if (curJob == null) {
    //        response.StatusMessage = ResponseStatus.ProcessJobResult_JobDoesNotExist;
    //        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
    //        Logger.Error("Job does not exist in DB: " + jobProgress.Key);
    //        return;
    //      }
    //      curJob.Slave = DaoLocator.SlaveDao.GetSlaveForJob(curJob.Id);
    //      if (curJob.Slave == null || curJob.Slave.Id != heartbeatData.SlaveId) {
    //        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
    //        Logger.Error("There is no job calculated by this user " + heartbeatData.SlaveId + " Job: " + curJob);
    //      } else if (curJob.State == JobState.Aborted) {
    //        // a request to abort the job has been set
    //        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
    //        curJob.State = JobState.Finished;
    //      } else {
    //        // save job progress
    //        curJob.ExecutionTime = jobProgress.Value;

    //        if (curJob.State == JobState.SnapshotRequested) {
    //          // a request for a snapshot has been set
    //          response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.RequestSnapshot, curJob.Id));
    //          curJob.State = JobState.SnapshotSent;
    //        }
    //      }
    //      DaoLocator.JobDao.Update(curJob);
    //    }
    //  }

    //  foreach (JobDto currJob in jobsOfSlave) {
    //    bool found = false;
    //    if (heartbeatData.JobProgress != null) {
    //      foreach (Guid jobId in heartbeatData.JobProgress.Keys) {
    //        if (jobId == currJob.Id) {
    //          found = true;
    //          break;
    //        }
    //      }
    //    }

    //    if (!found) {
    //      lock (newAssignedJobs) {
    //        if (newAssignedJobs.ContainsKey(currJob.Id)) {
    //          newAssignedJobs[currJob.Id]--;
    //          Logger.Error("Job TTL Reduced by one for job: " + currJob + "and is now: " + newAssignedJobs[currJob.Id] + ". User that sucks: " + currJob.Slave);
    //          if (newAssignedJobs[currJob.Id] <= 0) {
    //            Logger.Error("Job TTL reached Zero, Job gets removed: " + currJob + " and set back to offline. User that sucks: " + currJob.Slave);

    //            currJob.State = JobState.Offline;
    //            DaoLocator.JobDao.Update(currJob);

    //            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, currJob.Id));

    //            newAssignedJobs.Remove(currJob.Id);
    //          }
    //        } else {
    //          Logger.Error("Job ID wasn't with the heartbeats:  " + currJob);
    //          currJob.State = JobState.Offline;
    //          DaoLocator.JobDao.Update(currJob);
    //        }
    //      } // lock
    //    } else {
    //      lock (newAssignedJobs) {
    //        if (newAssignedJobs.ContainsKey(currJob.Id)) {
    //          Logger.Info("Job is sending a heart beat, removing it from the newAssignedJobList: " + currJob);
    //          newAssignedJobs.Remove(currJob.Id);
    //        }
    //      }
    //    }
    //  }
    //}

    /// <summary>
    /// if the slave was told to pull a job he calls this method
    /// the server selects a job and sends it to the slave
    /// </summary>
    /// <param name="slaveId"></param>
    /// <returns></returns>
    public ResponseObject<JobDto> GetJob(Guid slaveId) {
      ResponseObject<JobDto> response = new ResponseObject<JobDto>();

      JobDto job2Calculate = scheduler.GetNextJobForSlave(slaveId);
      if (job2Calculate != null) {
        response.Obj = job2Calculate;
        response.Obj.PluginsNeeded = DaoLocator.PluginInfoDao.GetPluginDependenciesForJob(response.Obj);

        Logger.Info("Job pulled: " + job2Calculate + " for user " + slaveId);
        job2Calculate.LastHeartbeat = DateTime.Now.AddSeconds(ApplicationConstants.JOB_TIME_TO_LIVE_SECONDS);
        DaoLocator.JobDao.Update(job2Calculate);
      } else {
        //response.Success = false;
        response.Obj = null;
        response.StatusMessage = ResponseStatus.GetJob_NoJobsAvailable;
        Logger.Info("No more Jobs left for " + slaveId);
      }
      return response;
    }

    public ResponseResultReceived ProcessJobResult(Stream stream, bool finished) {
      Logger.Info("BEGIN Job received for Storage - main method:");

      //Stream jobResultStream = null;
      //Stream jobStream = null;

      //try {
      BinaryFormatter formatter = new BinaryFormatter();

      JobResult result = (JobResult)formatter.Deserialize(stream);

      //important - repeatable read isolation level is required here, 
      //otherwise race conditions could occur when writing the stream into the DB
      //just removed TransactionIsolationLevel.RepeatableRead
      //tx = session.BeginTransaction();

      ResponseResultReceived response = ProcessJobResult(result.SlaveId, result.Id, new byte[] { }, result.ExecutionTime, result.Exception, finished);

      if (response.StatusMessage == ResponseStatus.Ok) {
        Logger.Debug("Trying to aquire WCF Job Stream");
        //jobStream = DaoLocator.JobDao.GetSerializedJobStream(result.JobId);
        //Logger.Debug("Job Stream Aquired");
        byte[] buffer = new byte[3024];
        List<byte> serializedJob = new List<byte>();
        int read = 0;
        int i = 0;
        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
          for (int j = 0; j < read; j++) {
            serializedJob.Add(buffer[j]);
          }
          if (i % 100 == 0)
            Logger.Debug("Writing to stream: " + i);
          //jobStream.Write(buffer, 0, read);
          i++;
        }
        Logger.Debug("Done Writing, closing the stream!");
        //jobStream.Close();

        if (serializedJob.Count > 0) {
          DaoLocator.JobDao.SetBinaryJobFile(result.Id, serializedJob.ToArray());
        }
      }
      Logger.Info("END Job received for Storage:");
      stream.Dispose();
      return response;
    }

    private ResponseResultReceived ProcessJobResult(Guid slaveId, Guid jobId, byte[] result, TimeSpan executionTime, string exception, bool finished) {
      Logger.Info("BEGIN Job received for Storage - SUB method: " + jobId);

      ResponseResultReceived response = new ResponseResultReceived();
      SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);

      SerializedJob job = new SerializedJob();

      if (job != null) {
        job.JobInfo = DaoLocator.JobDao.FindById(jobId);
        if (job.JobInfo != null) {
          job.JobInfo.Slave = job.JobInfo.Slave = DaoLocator.SlaveDao.GetSlaveForJob(jobId);
        }
      }

      if (job != null && job.JobInfo == null) {
        response.StatusMessage = ResponseStatus.ProcessJobResult_JobDoesNotExist;
        response.JobId = jobId;
        Logger.Error("No job with Id " + jobId);

        //tx.Rollback();
        return response;
      }
      if (job.JobInfo.State == JobState.Aborted) {
        response.StatusMessage = ResponseStatus.ProcessJobResult_JobAborted;

        Logger.Error("Job was aborted! " + job.JobInfo);

        //tx.Rollback();
        return response;
      }
      if (job.JobInfo.Slave == null) {
        response.StatusMessage = ResponseStatus.ProcessJobResult_JobIsNotBeeingCalculated;
        response.JobId = jobId;

        Logger.Error("Job is not being calculated (slave = null)! " + job.JobInfo);

        //tx.Rollback();
        return response;
      }
      if (job.JobInfo.Slave.Id != slaveId) {
        //response.Success = false;
        response.StatusMessage = ResponseStatus.ProcessJobResult_WrongSlaveForJob;
        response.JobId = jobId;

        Logger.Error("Wrong Slave for this Job! " + job.JobInfo + ", Sending Slave is: " + slaveId);

        //tx.Rollback();
        return response;
      }
      if (job.JobInfo.State == JobState.Finished) {
        response.StatusMessage = ResponseStatus.Ok;
        response.JobId = jobId;

        Logger.Error("Job already finished! " + job.JobInfo + ", Sending Slave is: " + slaveId);

        //tx.Rollback();
        return response;
      }
      //Todo: RequestsnapshotSent => calculating?
      if (job.JobInfo.State == JobState.SnapshotSent) {
        job.JobInfo.State = JobState.Calculating;
      }
      if (job.JobInfo.State != JobState.Calculating && job.JobInfo.State != JobState.Pending) {
        response.StatusMessage = ResponseStatus.ProcessJobResult_InvalidJobState;
        response.JobId = jobId;

        Logger.Error("Wrong Job State, job is: " + job.JobInfo);

        //tx.Rollback();
        return response;
      }
      job.JobInfo.ExecutionTime = executionTime;

      if (!string.IsNullOrEmpty(exception)) {
        job.JobInfo.State = JobState.Failed;
        job.JobInfo.Exception = exception;
        job.JobInfo.DateFinished = DateTime.Now;
      } else if (finished) {
        job.JobInfo.State = JobState.Finished;
        job.JobInfo.DateFinished = DateTime.Now;
      }

      job.SerializedJobData = result;

      DaoLocator.JobDao.Update(job.JobInfo);

      response.StatusMessage = ResponseStatus.Ok;
      response.JobId = jobId;
      response.Finished = finished;

      Logger.Info("END Job received for Storage - SUB method: " + jobId);
      return response;

    }

    /// <summary>
    /// the slave can send job results during calculating 
    /// and will send a final job result when he finished calculating
    /// these job results will be stored in the database
    /// </summary>
    public ResponseResultReceived StoreFinishedJobResult(Guid slaveId, Guid jobId, byte[] result, TimeSpan executionTime, string exception) {
      return ProcessJobResult(slaveId, jobId, result, executionTime, exception, true);
    }

    public ResponseResultReceived ProcessSnapshot(Guid slaveId, Guid jobId, byte[] result, TimeSpan executionTime, string exception) {
      return ProcessJobResult(slaveId, jobId, result, executionTime, exception, false);
    }

    /// <summary>
    /// when a slave logs out the state will be set
    /// and the entry in the last hearbeats dictionary will be removed
    /// </summary>
    /// <param name="slaveId"></param>
    /// <returns></returns>                       
    public Response Logout(Guid slaveId) {
      Logger.Info("Slave logged out " + slaveId);

      Response response = new Response();

      SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);
      if (slave == null) {
        response.StatusMessage = ResponseStatus.Logout_SlaveNotRegistered;
        return response;
      }
      if (slave.State == SlaveState.Calculating) {
        // check wich job the slave was calculating and reset it
        IEnumerable<JobDto> jobsOfSlave = DaoLocator.JobDao.FindActiveJobsOfSlave(slave);
        foreach (JobDto job in jobsOfSlave) {
          if (job.State != JobState.Finished)
            DaoLocator.JobDao.SetJobOffline(job);
        }
      }

      slave.State = SlaveState.Offline;
      DaoLocator.SlaveDao.Update(slave);

      return response;
    }

    /// <summary>
    /// If a slave goes offline and restores a job he was calculating 
    /// he can ask the slave if he still needs the job result
    /// </summary>
    /// <param name="jobId"></param>
    /// <returns></returns>
    /// TODO: Just reset LastHeartbeat timestamp in this method
    public Response IsJobStillNeeded(Guid jobId) {
      Response response = new Response();
      JobDto job = DaoLocator.JobDao.FindById(jobId);
      if (job == null) {
        response.StatusMessage = ResponseStatus.IsJobStillNeeded_JobDoesNotExist;
        Logger.Error("Job doesn't exist (anymore)! " + jobId);
        return response;
      }
      if (job.State == JobState.Finished) {
        response.StatusMessage = ResponseStatus.IsJobStillNeeded_JobAlreadyFinished;
        Logger.Error("already finished! " + job);
        return response;
      }
      job.State = JobState.Pending;
      //lock (lifecycleManager.PendingJobs) {
      //  lifecycleManager.PendingJobs.Add(job.Id, PENDING_TIMEOUT);
      //}

      DaoLocator.JobDao.Update(job);

      return response;
    }

    public ResponseList<CachedHivePluginInfoDto> GetPlugins(List<HivePluginInfoDto> pluginList) {
      ResponseList<CachedHivePluginInfoDto> response = new ResponseList<CachedHivePluginInfoDto>();
      response.List = new List<CachedHivePluginInfoDto>();
      foreach (HivePluginInfoDto pluginInfo in pluginList) {
        IPluginDescription ipd = GetAvailablePlugins().Where(pd =>
          pd.Name == pluginInfo.Name &&
          pd.Version.Major == pluginInfo.Version.Major &&
          pd.Version.Minor == pluginInfo.Version.Minor &&
          pd.Version.Build >= pluginInfo.Version.Build).SingleOrDefault(); // ignore revision here, otherwise each client who uses a more recent trunk-build than the server-build will not be able to execute the job
        if (ipd != null) {
          response.List.Add(ConvertPluginDescriptorToDto(ipd));
        }
      }
      return response;
    }

    private IEnumerable<PluginDescription> GetAvailablePlugins() {
      return lifecycleManager.Plugins;
    }

    private CachedHivePluginInfoDto ConvertPluginDescriptorToDto(IPluginDescription currPlugin) {
      CachedHivePluginInfoDto currCachedPlugin = new CachedHivePluginInfoDto {
        Name = currPlugin.Name,
        Version = currPlugin.Version
      };

      foreach (string fileName in from file in currPlugin.Files select file.Name) {
        currCachedPlugin.PluginFiles.Add(new HivePluginFile(File.ReadAllBytes(fileName), fileName));
      }
      return currCachedPlugin;
    }

    public ResponseObject<HivePluginFile> GetConfigurationFile() {
      var response = new ResponseObject<HivePluginFile>();
      response.Obj = new HivePluginFile(lifecycleManager.ConfigurationFile, "Sandbox.config");
      return response;
    }

    #endregion
  }
}