source: branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/ClientCommunicator.cs @ 4173

Last change on this file since 4173 was 4173, checked in by cneumuel, 12 years ago
  • reorganized HiveExperiment code
  • disabled snapshot-functionality... this needs more refactoring serverside
  • added short documentation which explains how to use hive
  • some minor changes
File size: 30.5 KB
Line 
1#region License Information
2/* HeuristicLab
3 * Copyright (C) 2002-2008 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
4 *
5 * This file is part of HeuristicLab.
6 *
7 * HeuristicLab is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * HeuristicLab is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
19 */
20#endregion
21
22using System;
23using System.Collections.Generic;
24using System.IO;
25using System.Linq;
26using System.Runtime.Serialization.Formatters.Binary;
27using System.Threading;
28using System.Transactions;
29using HeuristicLab.Hive.Contracts;
30using HeuristicLab.Hive.Contracts.BusinessObjects;
31using HeuristicLab.Hive.Contracts.Interfaces;
32using HeuristicLab.Hive.Server.Core.InternalInterfaces;
33using HeuristicLab.PluginInfrastructure;
34using HeuristicLab.Tracing;
35
36namespace HeuristicLab.Hive.Server.Core {
37  /// <summary>
38  /// The ClientCommunicator manages the whole communication with the client
39  /// </summary>
40  public class ClientCommunicator : IClientCommunicator,
41    IInternalClientCommunicator {
42    private static Dictionary<Guid, DateTime> lastHeartbeats = new Dictionary<Guid, DateTime>();
43    private static Dictionary<Guid, int> newAssignedJobs = new Dictionary<Guid, int>();
44    private static Dictionary<Guid, int> pendingJobs = new Dictionary<Guid, int>();
45
46    private static ReaderWriterLockSlim heartbeatLock = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
47
48    //private ISessionFactory factory;
49    private ILifecycleManager lifecycleManager;
50    private IInternalJobManager jobManager;
51    private IScheduler scheduler;
52
53    private static int PENDING_TIMEOUT = 100;
54
55    /// <summary>
56    /// Initialization of the Adapters to the database
57    /// Initialization of Eventhandler for the lifecycle management
58    /// Initialization of lastHearbeats Dictionary
59    /// </summary>
60    public ClientCommunicator() {
61      //factory = ServiceLocator.GetSessionFactory();
62
63      lifecycleManager = ServiceLocator.GetLifecycleManager();
64      jobManager = ServiceLocator.GetJobManager() as IInternalJobManager;
65      scheduler = ServiceLocator.GetScheduler();
66
67      lifecycleManager.RegisterHeartbeat(new EventHandler(lifecycleManager_OnServerHeartbeat));
68    }
69
70    /// <summary>
71    /// Check if online clients send their hearbeats
72    /// if not -> set them offline and check if they where calculating a job
73    /// </summary>
74    /// <param name="sender"></param>
75    /// <param name="e"></param>
76    void lifecycleManager_OnServerHeartbeat(object sender, EventArgs e) {
77      Logger.Debug("Server Heartbeat ticked");
78
79      // [chn] why is transaction management done here
80      using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = ApplicationConstants.ISOLATION_LEVEL_SCOPE })) {
81        List<ClientDto> allClients = new List<ClientDto>(DaoLocator.ClientDao.FindAll());
82
83        foreach (ClientDto client in allClients) {
84          if (client.State != State.Offline && client.State != State.NullState) {
85            heartbeatLock.EnterUpgradeableReadLock();
86
87            if (!lastHeartbeats.ContainsKey(client.Id)) {
88              Logger.Info("Client " + client.Id +
89                              " wasn't offline but hasn't sent heartbeats - setting offline");
90              client.State = State.Offline;
91              DaoLocator.ClientDao.Update(client);
92              Logger.Info("Client " + client.Id +
93                              " wasn't offline but hasn't sent heartbeats - Resetting all his jobs");
94              foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfClient(client)) {
95                //maybe implementa n additional Watchdog? Till then, just set them offline..
96                DaoLocator.JobDao.SetJobOffline(job);
97              }
98            } else {
99              DateTime lastHbOfClient = lastHeartbeats[client.Id];
100
101              TimeSpan dif = DateTime.Now.Subtract(lastHbOfClient);
102              // check if time between last hearbeat and now is greather than HEARTBEAT_MAX_DIF
103              if (dif.TotalSeconds > ApplicationConstants.HEARTBEAT_MAX_DIF) {
104                // if client calculated jobs, the job must be reset
105                Logger.Info("Client timed out and is on RESET");
106                foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfClient(client)) {
107                  DaoLocator.JobDao.SetJobOffline(job);
108                  lock (newAssignedJobs) {
109                    if (newAssignedJobs.ContainsKey(job.Id))
110                      newAssignedJobs.Remove(job.Id);
111                  }
112                }
113                Logger.Debug("setting client offline");
114                // client must be set offline
115                client.State = State.Offline;
116
117                //clientAdapter.Update(client);
118                DaoLocator.ClientDao.Update(client);
119
120                Logger.Debug("removing it from the heartbeats list");
121                heartbeatLock.EnterWriteLock();
122                lastHeartbeats.Remove(client.Id);
123                heartbeatLock.ExitWriteLock();
124              }
125            }
126
127            heartbeatLock.ExitUpgradeableReadLock();
128          } else {
129            //TODO: RLY neccesary?
130            //HiveLogger.Info(this.ToString() + ": Client " + client.Id + " has wrong state: Shouldn't have offline or nullstate, has " + client.State);
131            heartbeatLock.EnterWriteLock();
132            //HiveLogger.Info(this.ToString() + ": Client " + client.Id + " has wrong state: Resetting all his jobs");
133            if (lastHeartbeats.ContainsKey(client.Id))
134              lastHeartbeats.Remove(client.Id);
135            foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfClient(client)) {
136              DaoLocator.JobDao.SetJobOffline(job);
137            }
138            heartbeatLock.ExitWriteLock();
139          }
140        }
141        CheckForPendingJobs();
142        //        DaoLocator.DestroyContext();
143        scope.Complete();
144      }
145    }
146
147    private void CheckForPendingJobs() {
148      IList<JobDto> pendingJobsInDB = new List<JobDto>(DaoLocator.JobDao.GetJobsByState(State.Pending));
149
150      foreach (JobDto currJob in pendingJobsInDB) {
151        lock (pendingJobs) {
152          if (pendingJobs.ContainsKey(currJob.Id)) {
153            if (pendingJobs[currJob.Id] <= 0) {
154              currJob.State = State.Offline;
155              DaoLocator.JobDao.Update(currJob);
156            } else {
157              pendingJobs[currJob.Id]--;
158            }
159          }
160        }
161      }
162    }
163
164    #region IClientCommunicator Members
165
166    /// <summary>
167    /// Login process for the client
168    /// A hearbeat entry is created as well (login is the first hearbeat)
169    /// </summary>
170    /// <param name="clientInfo"></param>
171    /// <returns></returns>
172    public Response Login(ClientDto clientInfo) {
173      Response response = new Response();
174
175      heartbeatLock.EnterWriteLock();
176      if (lastHeartbeats.ContainsKey(clientInfo.Id)) {
177        lastHeartbeats[clientInfo.Id] = DateTime.Now;
178      } else {
179        lastHeartbeats.Add(clientInfo.Id, DateTime.Now);
180      }
181      heartbeatLock.ExitWriteLock();
182
183      ClientDto dbClient = DaoLocator.ClientDao.FindById(clientInfo.Id);
184
185      //Really set offline?
186      //Reconnect issues with the currently calculating jobs
187      clientInfo.State = State.Idle;
188      clientInfo.CalendarSyncStatus = dbClient != null ? dbClient.CalendarSyncStatus : CalendarState.NotAllowedToFetch;
189
190      if (dbClient == null)
191        DaoLocator.ClientDao.Insert(clientInfo);
192      else
193        DaoLocator.ClientDao.Update(clientInfo);
194      response.Success = true;
195      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_SUCCESS;
196      return response;
197    }
198
199    public ResponseCalendar GetCalendar(Guid clientId) {
200      ResponseCalendar response = new ResponseCalendar();
201
202      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
203      if (client == null) {
204        response.Success = false;
205        response.StatusMessage = ApplicationConstants.RESPONSE_CLIENT_RESOURCE_NOT_FOUND;
206        return response;
207      }
208
209      response.ForceFetch = (client.CalendarSyncStatus == CalendarState.ForceFetch);
210
211      IEnumerable<AppointmentDto> appointments = DaoLocator.UptimeCalendarDao.GetCalendarForClient(client);
212      if (appointments.Count() == 0) {
213        response.StatusMessage = ApplicationConstants.RESPONSE_UPTIMECALENDAR_NO_CALENDAR_FOUND;
214        response.Success = false;
215      } else {
216        response.Success = true;
217        response.Appointments = appointments;
218      }
219
220      client.CalendarSyncStatus = CalendarState.Fetched;
221      DaoLocator.ClientDao.Update(client);
222      return response;
223    }
224
225    public Response SetCalendarStatus(Guid clientId, CalendarState state) {
226      Response response = new Response();
227      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
228      if (client == null) {
229        response.Success = false;
230        response.StatusMessage = ApplicationConstants.RESPONSE_CLIENT_RESOURCE_NOT_FOUND;
231        return response;
232      }
233
234      client.CalendarSyncStatus = state;
235      DaoLocator.ClientDao.Update(client);
236
237      response.Success = true;
238      response.StatusMessage = ApplicationConstants.RESPONSE_UPTIMECALENDAR_STATUS_UPDATED;
239
240      return response;
241    }
242
243    /// <summary>
244    /// The client has to send regulary heartbeats
245    /// this hearbeats will be stored in the heartbeats dictionary
246    /// check if there is work for the client and send the client a response if he should pull a job
247    /// </summary>
248    /// <param name="hbData"></param>
249    /// <returns></returns>
250    public ResponseHB ProcessHeartBeat(HeartBeatData hbData) {
251      Logger.Debug("BEGIN Processing Heartbeat for Client " + hbData.ClientId);
252
253      ResponseHB response = new ResponseHB();
254      response.ActionRequest = new List<MessageContainer>();
255
256      Logger.Debug("BEGIN Started Client Fetching");
257      ClientDto client = DaoLocator.ClientDao.FindById(hbData.ClientId);
258      Logger.Debug("END Finished Client Fetching");
259      // check if the client is logged in
260      if (client.State == State.Offline || client.State == State.NullState) {
261        response.Success = false;
262        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_USER_NOT_LOGGED_IN;
263        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
264
265        Logger.Error("ProcessHeartBeat: Client state null or offline: " + client);
266
267        return response;
268      }
269
270      client.NrOfFreeCores = hbData.FreeCores;
271      client.FreeMemory = hbData.FreeMemory;
272
273      // save timestamp of this heartbeat
274      Logger.Debug("BEGIN Locking for Heartbeats");
275      heartbeatLock.EnterWriteLock();
276      Logger.Debug("END Locked for Heartbeats");
277      if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
278        lastHeartbeats[hbData.ClientId] = DateTime.Now;
279      } else {
280        lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
281      }
282      heartbeatLock.ExitWriteLock();
283
284      Logger.Debug("BEGIN Processing Heartbeat Jobs");
285      ProcessJobProcess(hbData, response);
286      Logger.Debug("END Processed Heartbeat Jobs");
287
288      //check if new Cal must be loaded
289      if (client.CalendarSyncStatus == CalendarState.Fetch || client.CalendarSyncStatus == CalendarState.ForceFetch) {
290        response.Success = true;
291        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_FETCH_OR_FORCEFETCH_CALENDAR;
292        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchOrForceFetchCalendar));
293
294        //client.CalendarSyncStatus = CalendarState.Fetching;
295
296        Logger.Info("fetch or forcefetch sent");
297      }
298
299      // check if client has a free core for a new job
300      // if true, ask scheduler for a new job for this client
301      Logger.Debug(" BEGIN Looking for Client Jobs");
302      if (hbData.FreeCores > 0 && scheduler.ExistsJobForClient(hbData)) {
303        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
304      } else {
305        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
306      }
307      Logger.Debug(" END Looked for Client Jobs");
308      response.Success = true;
309      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_HEARTBEAT_RECEIVED;
310
311      DaoLocator.ClientDao.Update(client);
312
313      //tx.Commit();
314      Logger.Debug(" END Processed Heartbeat for Client " + hbData.ClientId);
315      return response;
316    }
317
318    /// <summary>
319    /// Process the Job progress sent by a client
320    /// [chn] this method needs to be refactored, because its a performance hog
321    ///
322    /// what it does:
323    /// (1) find out if the jobs that should be calculated by this client (from db) and compare if they are consistent with what the joblist the client sent
324    /// (2) find out if every job from the joblist really should be calculated by this client
325    /// (3) checks if a job should be aborted and issues Message
326    /// (4) update job-progress and write to db
327    /// (5) if snapshot is requested, issue Message
328    ///
329    /// (6) for each job from DB, check if there is a job from client (again).
330    /// (7) if job matches, it is removed from newAssigneJobs
331    /// (8) if job !matches, job's TTL is reduced by 1,
332    /// (9) if TTL==0, job is set to Abort (save to DB), and Message to Abort job is issued to client
333    ///
334    ///
335    ///
336    /// quirks:
337    /// (1) the response-object is modified during the foreach-loop (only last element counts)
338    /// (2) state Abort results in Finished. This should be: AbortRequested, Aborted.
339    /// </summary>
340    /// <param name="hbData"></param>
341    /// <param name="jobAdapter"></param>
342    /// <param name="clientAdapter"></param>
343    /// <param name="response"></param>
344    private void ProcessJobProcess(HeartBeatData hbData, ResponseHB response) {
345      Logger.Debug("Started for Client " + hbData.ClientId);
346      List<JobDto> jobsOfClient = new List<JobDto>(DaoLocator.JobDao.FindActiveJobsOfClient(DaoLocator.ClientDao.FindById(hbData.ClientId)));
347      if (hbData.JobProgress != null && hbData.JobProgress.Count > 0) {
348        if (jobsOfClient == null || jobsOfClient.Count == 0) {
349          response.Success = false;
350          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
351
352          foreach (Guid jobId in hbData.JobProgress.Keys) {
353            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobId));
354          }
355
356          Logger.Error("There is no job calculated by this user " + hbData.ClientId + ", advise him to abort all");
357          return;
358        }
359
360        foreach (KeyValuePair<Guid, double> jobProgress in hbData.JobProgress) {
361          JobDto curJob = DaoLocator.JobDao.FindById(jobProgress.Key);
362          curJob.Client = DaoLocator.ClientDao.GetClientForJob(curJob.Id);
363          if (curJob.Client == null || curJob.Client.Id != hbData.ClientId) {
364            response.Success = false;
365            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
366            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
367            Logger.Error("There is no job calculated by this user " + hbData.ClientId + " Job: " + curJob);
368          } else if (curJob.State == State.Abort) {
369            // a request to abort the job has been set
370            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
371            curJob.State = State.Finished;
372          } else {
373            // save job progress
374            curJob.Percentage = jobProgress.Value;
375
376            if (curJob.State == State.RequestSnapshot) {
377              // a request for a snapshot has been set
378              response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.RequestSnapshot, curJob.Id));
379              curJob.State = State.RequestSnapshotSent;
380            }
381          }
382          DaoLocator.JobDao.Update(curJob);
383        }
384      }
385      foreach (JobDto currJob in jobsOfClient) {
386        bool found = false;
387        if (hbData.JobProgress != null) {
388          foreach (Guid jobId in hbData.JobProgress.Keys) {
389            if (jobId == currJob.Id) {
390              found = true;
391              break;
392            }
393          }
394        }
395        if (!found) {
396          lock (newAssignedJobs) {
397            if (newAssignedJobs.ContainsKey(currJob.Id)) {
398              newAssignedJobs[currJob.Id]--;
399              Logger.Error("Job TTL Reduced by one for job: " + currJob + "and is now: " + newAssignedJobs[currJob.Id] + ". User that sucks: " + currJob.Client);
400              if (newAssignedJobs[currJob.Id] <= 0) {
401                Logger.Error("Job TTL reached Zero, Job gets removed: " + currJob + " and set back to offline. User that sucks: " + currJob.Client);
402
403                currJob.State = State.Offline;
404                DaoLocator.JobDao.Update(currJob);
405
406                response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, currJob.Id));
407
408                newAssignedJobs.Remove(currJob.Id);
409              }
410            } else {
411              Logger.Error("Job ID wasn't with the heartbeats:  " + currJob);
412              currJob.State = State.Offline;
413              DaoLocator.JobDao.Update(currJob);
414            }
415          } // lock
416        } else {
417          lock (newAssignedJobs) {
418
419            if (newAssignedJobs.ContainsKey(currJob.Id)) {
420              Logger.Info("Job is sending a heart beat, removing it from the newAssignedJobList: " + currJob);
421              newAssignedJobs.Remove(currJob.Id);
422            }
423          }
424        }
425      }
426    }
427
428    /// <summary>
429    /// if the client was told to pull a job he calls this method
430    /// the server selects a job and sends it to the client
431    /// </summary>
432    /// <param name="clientId"></param>
433    /// <returns></returns>
434    public ResponseJob SendJob(Guid clientId) {
435
436      ResponseJob response = new ResponseJob();
437
438      JobDto job2Calculate = scheduler.GetNextJobForClient(clientId);
439      if (job2Calculate != null) {
440        response.Job = job2Calculate;
441        response.Job.PluginsNeeded = DaoLocator.PluginInfoDao.GetPluginDependenciesForJob(response.Job);
442        response.Success = true;
443        Logger.Info("Job pulled: " + job2Calculate + " for user " + clientId);
444        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
445        lock (newAssignedJobs) {
446          if (!newAssignedJobs.ContainsKey(job2Calculate.Id))
447            newAssignedJobs.Add(job2Calculate.Id, ApplicationConstants.JOB_TIME_TO_LIVE);
448        }
449      } else {
450        response.Success = false;
451        response.Job = null;
452        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
453        Logger.Info("No more Jobs left for " + clientId);
454      }
455
456
457
458      return response;
459    }
460
461    public ResponseResultReceived ProcessJobResult(Stream stream, bool finished) {
462      Logger.Info("BEGIN Job received for Storage - main method:");
463
464      //Stream jobResultStream = null;
465      //Stream jobStream = null;
466
467      //try {
468      BinaryFormatter formatter = new BinaryFormatter();
469
470      JobResult result = (JobResult)formatter.Deserialize(stream);
471
472      //important - repeatable read isolation level is required here,
473      //otherwise race conditions could occur when writing the stream into the DB
474      //just removed TransactionIsolationLevel.RepeatableRead
475      //tx = session.BeginTransaction();
476
477      ResponseResultReceived response = ProcessJobResult(result.ClientId, result.JobId, new byte[] { }, result.Percentage, result.Exception, finished);
478
479      if (response.Success) {
480        Logger.Debug("Trying to aquire WCF Job Stream");
481        //jobStream = DaoLocator.JobDao.GetSerializedJobStream(result.JobId);
482        //Logger.Debug("Job Stream Aquired");
483        byte[] buffer = new byte[3024];
484        List<byte> serializedJob = new List<byte>();
485        int read = 0;
486        int i = 0;
487        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
488          for (int j = 0; j < read; j++) {
489            serializedJob.Add(buffer[j]);
490          }
491          if (i % 100 == 0)
492            Logger.Debug("Writing to stream: " + i);
493          //jobStream.Write(buffer, 0, read);
494          i++;
495        }
496        Logger.Debug("Done Writing, closing the stream!");
497        //jobStream.Close();
498
499        DaoLocator.JobDao.SetBinaryJobFile(result.JobId, serializedJob.ToArray());
500      }
501      Logger.Info("END Job received for Storage:");
502      stream.Dispose();
503      return response;
504    }
505
506
507    private ResponseResultReceived ProcessJobResult(Guid clientId,
508      Guid jobId,
509      byte[] result,
510      double? percentage,
511      string exception,
512      bool finished) {
513
514      Logger.Info("BEGIN Job received for Storage - SUB method: " + jobId);
515
516      ResponseResultReceived response = new ResponseResultReceived();
517      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
518
519      SerializedJob job = new SerializedJob();
520
521      if (job != null) {
522        job.JobInfo = DaoLocator.JobDao.FindById(jobId);
523        if (job.JobInfo != null) {
524          job.JobInfo.Client = job.JobInfo.Client = DaoLocator.ClientDao.GetClientForJob(jobId);
525        }
526      }
527     
528      if (job != null && job.JobInfo == null) {
529        response.Success = false;
530        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
531        response.JobId = jobId;
532        Logger.Error("No job with Id " + jobId);
533
534        //tx.Rollback();
535        return response;
536      }
537      if (job.JobInfo.State == State.Abort) {
538        response.Success = false;
539        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_WAS_ABORTED;
540
541        Logger.Error("Job was aborted! " + job.JobInfo);
542
543        //tx.Rollback();
544        return response;
545      }
546      if (job.JobInfo.Client == null) {
547        response.Success = false;
548        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
549        response.JobId = jobId;
550
551        Logger.Error("Job is not being calculated (client = null)! " + job.JobInfo);
552
553        //tx.Rollback();
554        return response;
555      }
556      if (job.JobInfo.Client.Id != clientId) {
557        response.Success = false;
558        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
559        response.JobId = jobId;
560
561        Logger.Error("Wrong Client for this Job! " + job.JobInfo + ", Sending Client is: " + clientId);
562
563        //tx.Rollback();
564        return response;
565      }
566      if (job.JobInfo.State == State.Finished) {
567        response.Success = true;
568        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
569        response.JobId = jobId;
570
571        Logger.Error("Job already finished! " + job.JobInfo + ", Sending Client is: " + clientId);
572
573        //tx.Rollback();
574        return response;
575      }
576      //Todo: RequestsnapshotSent => calculating?
577      if (job.JobInfo.State == State.RequestSnapshotSent) {
578        job.JobInfo.State = State.Calculating;
579      }
580      if (job.JobInfo.State != State.Calculating &&
581        job.JobInfo.State != State.Pending) {
582        response.Success = false;
583        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
584        response.JobId = jobId;
585
586        Logger.Error("Wrong Job State, job is: " + job.JobInfo);
587
588        //tx.Rollback();
589        return response;
590      }
591      job.JobInfo.Percentage = percentage;
592
593      if (!string.IsNullOrEmpty(exception)) {
594        job.JobInfo.State = State.Failed;
595        job.JobInfo.Exception = exception;
596        job.JobInfo.DateFinished = DateTime.Now;
597      } else if (finished) {
598        job.JobInfo.State = State.Finished;
599        job.JobInfo.DateFinished = DateTime.Now;
600      }
601
602      job.SerializedJobData = result;
603
604      DaoLocator.JobDao.Update(job.JobInfo);
605
606      response.Success = true;
607      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
608      response.JobId = jobId;
609      response.Finished = finished;
610
611      Logger.Info("END Job received for Storage - SUB method: " + jobId);
612      return response;
613
614    }
615
616    /// <summary>
617    /// the client can send job results during calculating
618    /// and will send a final job result when he finished calculating
619    /// these job results will be stored in the database
620    /// </summary>
621    /// <param name="clientId"></param>
622    /// <param name="jobId"></param>
623    /// <param name="result"></param>
624    /// <param name="exception"></param>
625    /// <param name="finished"></param>
626    /// <returns></returns>
627    public ResponseResultReceived StoreFinishedJobResult(Guid clientId,
628      Guid jobId,
629      byte[] result,
630      double percentage,
631      string exception) {
632
633      return ProcessJobResult(clientId, jobId, result, percentage, exception, true);
634    }
635
636    public ResponseResultReceived ProcessSnapshot(Guid clientId, Guid jobId, byte[] result, double percentage, string exception) {
637      return ProcessJobResult(clientId, jobId, result, percentage, exception, false);
638    }
639
640    /// <summary>
641    /// when a client logs out the state will be set
642    /// and the entry in the last hearbeats dictionary will be removed
643    /// </summary>
644    /// <param name="clientId"></param>
645    /// <returns></returns>                       
646    public Response Logout(Guid clientId) {
647      Logger.Info("Client logged out " + clientId);
648
649      Response response = new Response();
650
651      heartbeatLock.EnterWriteLock();
652      if (lastHeartbeats.ContainsKey(clientId))
653        lastHeartbeats.Remove(clientId);
654      heartbeatLock.ExitWriteLock();
655
656      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
657      if (client == null) {
658        response.Success = false;
659        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
660        return response;
661      }
662      if (client.State == State.Calculating) {
663        // check wich job the client was calculating and reset it
664        IEnumerable<JobDto> jobsOfClient = DaoLocator.JobDao.FindActiveJobsOfClient(client);
665        foreach (JobDto job in jobsOfClient) {
666          if (job.State != State.Finished)
667            DaoLocator.JobDao.SetJobOffline(job);
668        }
669      }
670
671      client.State = State.Offline;
672      DaoLocator.ClientDao.Update(client);
673
674      response.Success = true;
675      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
676
677      return response;
678    }
679
680    /// <summary>
681    /// If a client goes offline and restores a job he was calculating
682    /// he can ask the client if he still needs the job result
683    /// </summary>
684    /// <param name="jobId"></param>
685    /// <returns></returns>
686    public Response IsJobStillNeeded(Guid jobId) {
687      Response response = new Response();
688      JobDto job = DaoLocator.JobDao.FindById(jobId);
689      if (job == null) {
690        response.Success = false;
691        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_DOESNT_EXIST;
692        Logger.Error("Job doesn't exist (anymore)! " + jobId);
693        return response;
694      }
695      if (job.State == State.Finished) {
696        response.Success = true;
697        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_ALLREADY_FINISHED;
698        Logger.Error("already finished! " + job);
699        return response;
700      }
701      job.State = State.Pending;
702      lock (pendingJobs) {
703        pendingJobs.Add(job.Id, PENDING_TIMEOUT);
704      }
705
706      DaoLocator.JobDao.Update(job);
707
708      response.Success = true;
709      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_SEND_JOBRESULT;
710      return response;
711    }
712
713    public ResponsePlugin SendPlugins(List<HivePluginInfoDto> pluginList) {
714      ResponsePlugin response = new ResponsePlugin();
715      foreach (HivePluginInfoDto pluginInfo in pluginList) {
716        if (pluginInfo.Update) {
717          //check if there is a newer version         
718          IPluginDescription ipd =
719            ApplicationManager.Manager.Plugins.Where(pd => pd.Name == pluginInfo.Name && pd.Version.Major == pluginInfo.Version.Major && pd.Version.Minor == pluginInfo.Version.Minor && pd.Version.Revision > pluginInfo.Version.Revision).SingleOrDefault();
720          if (ipd != null) {
721            response.Plugins.Add(convertPluginDescriptorToDto(ipd));
722          }
723        } else {
724          IPluginDescription ipd =
725            ApplicationManager.Manager.Plugins.Where(pd => pd.Name == pluginInfo.Name && pd.Version.Major == pluginInfo.Version.Major && pd.Version.Minor == pluginInfo.Version.Minor && pd.Version.Revision >= pluginInfo.Version.Revision).SingleOrDefault();
726          if (ipd != null) {
727            response.Plugins.Add(convertPluginDescriptorToDto(ipd));
728          } else {
729            response.Success = false;
730            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_PLUGINS_NOT_AVAIL;
731            return response;
732          }
733        }
734      }
735      response.Success = true;
736      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_PLUGINS_SENT;
737
738      return response;
739    }
740
741    private CachedHivePluginInfoDto convertPluginDescriptorToDto(IPluginDescription currPlugin) {
742      CachedHivePluginInfoDto currCachedPlugin = new CachedHivePluginInfoDto {
743        Name = currPlugin.Name,
744        Version = currPlugin.Version
745      };
746
747      foreach (string fileName in from file in currPlugin.Files select file.Name) {
748        currCachedPlugin.PluginFiles.Add(new HivePluginFile(File.ReadAllBytes(fileName), fileName));
749      }
750      return currCachedPlugin;
751    }
752
753    #endregion
754  }
755}
Note: See TracBrowser for help on using the repository browser.