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

Last change on this file since 4170 was 4170, checked in by cneumuel, 12 years ago

refactoring of Result-Polling of HiveExperiment, polling is now much faster and code is cleaner (1092#)

File size: 30.2 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          Logger.Error("There is no job calculated by this user " + hbData.ClientId);
352          return;
353        }
354
355        foreach (KeyValuePair<Guid, double> jobProgress in hbData.JobProgress) {
356          JobDto curJob = DaoLocator.JobDao.FindById(jobProgress.Key);
357          curJob.Client = DaoLocator.ClientDao.GetClientForJob(curJob.Id);
358          if (curJob.Client == null || curJob.Client.Id != hbData.ClientId) {
359            response.Success = false;
360            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
361            Logger.Error("There is no job calculated by this user " + hbData.ClientId + " Job: " + curJob);
362          } else if (curJob.State == State.Abort) {
363            // a request to abort the job has been set
364            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
365            curJob.State = State.Finished;
366          } else {
367            // save job progress
368            curJob.Percentage = jobProgress.Value;
369
370            if (curJob.State == State.RequestSnapshot) {
371              // a request for a snapshot has been set
372              response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.RequestSnapshot, curJob.Id));
373              curJob.State = State.RequestSnapshotSent;
374            }
375          }
376          DaoLocator.JobDao.Update(curJob);
377        }
378      }
379      foreach (JobDto currJob in jobsOfClient) {
380        bool found = false;
381        if (hbData.JobProgress != null) {
382          foreach (Guid jobId in hbData.JobProgress.Keys) {
383            if (jobId == currJob.Id) {
384              found = true;
385              break;
386            }
387          }
388        }
389        if (!found) {
390          lock (newAssignedJobs) {
391            if (newAssignedJobs.ContainsKey(currJob.Id)) {
392              newAssignedJobs[currJob.Id]--;
393              Logger.Error("Job TTL Reduced by one for job: " + currJob + "and is now: " + newAssignedJobs[currJob.Id] + ". User that sucks: " + currJob.Client);
394              if (newAssignedJobs[currJob.Id] <= 0) {
395                Logger.Error("Job TTL reached Zero, Job gets removed: " + currJob + " and set back to offline. User that sucks: " + currJob.Client);
396
397                currJob.State = State.Offline;
398                DaoLocator.JobDao.Update(currJob);
399
400                response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, currJob.Id));
401
402                newAssignedJobs.Remove(currJob.Id);
403              }
404            } else {
405              Logger.Error("Job ID wasn't with the heartbeats:  " + currJob);
406              currJob.State = State.Offline;
407              DaoLocator.JobDao.Update(currJob);
408            }
409          } // lock
410        } else {
411          lock (newAssignedJobs) {
412
413            if (newAssignedJobs.ContainsKey(currJob.Id)) {
414              Logger.Info("Job is sending a heart beat, removing it from the newAssignedJobList: " + currJob);
415              newAssignedJobs.Remove(currJob.Id);
416            }
417          }
418        }
419      }
420    }
421
422    /// <summary>
423    /// if the client was told to pull a job he calls this method
424    /// the server selects a job and sends it to the client
425    /// </summary>
426    /// <param name="clientId"></param>
427    /// <returns></returns>
428    public ResponseJob SendJob(Guid clientId) {
429
430      ResponseJob response = new ResponseJob();
431
432      JobDto job2Calculate = scheduler.GetNextJobForClient(clientId);
433      if (job2Calculate != null) {
434        response.Job = job2Calculate;
435        response.Job.PluginsNeeded = DaoLocator.PluginInfoDao.GetPluginDependenciesForJob(response.Job);
436        response.Success = true;
437        Logger.Info("Job pulled: " + job2Calculate + " for user " + clientId);
438        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
439        lock (newAssignedJobs) {
440          if (!newAssignedJobs.ContainsKey(job2Calculate.Id))
441            newAssignedJobs.Add(job2Calculate.Id, ApplicationConstants.JOB_TIME_TO_LIVE);
442        }
443      } else {
444        response.Success = false;
445        response.Job = null;
446        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
447        Logger.Info("No more Jobs left for " + clientId);
448      }
449
450
451
452      return response;
453    }
454
455    public ResponseResultReceived ProcessJobResult(Stream stream, bool finished) {
456      Logger.Info("BEGIN Job received for Storage - main method:");
457
458      //Stream jobResultStream = null;
459      //Stream jobStream = null;
460
461      //try {
462      BinaryFormatter formatter = new BinaryFormatter();
463
464      JobResult result = (JobResult)formatter.Deserialize(stream);
465
466      //important - repeatable read isolation level is required here,
467      //otherwise race conditions could occur when writing the stream into the DB
468      //just removed TransactionIsolationLevel.RepeatableRead
469      //tx = session.BeginTransaction();
470
471      ResponseResultReceived response = ProcessJobResult(result.ClientId, result.JobId, new byte[] { }, result.Percentage, result.Exception, finished);
472
473      if (response.Success) {
474        Logger.Debug("Trying to aquire WCF Job Stream");
475        //jobStream = DaoLocator.JobDao.GetSerializedJobStream(result.JobId);
476        //Logger.Debug("Job Stream Aquired");
477        byte[] buffer = new byte[3024];
478        List<byte> serializedJob = new List<byte>();
479        int read = 0;
480        int i = 0;
481        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
482          for (int j = 0; j < read; j++) {
483            serializedJob.Add(buffer[j]);
484          }
485          if (i % 100 == 0)
486            Logger.Debug("Writing to stream: " + i);
487          //jobStream.Write(buffer, 0, read);
488          i++;
489        }
490        Logger.Debug("Done Writing, closing the stream!");
491        //jobStream.Close();
492
493        DaoLocator.JobDao.SetBinaryJobFile(result.JobId, serializedJob.ToArray());
494      }
495      Logger.Info("END Job received for Storage:");
496      stream.Dispose();
497      return response;
498    }
499
500
501    private ResponseResultReceived ProcessJobResult(Guid clientId,
502      Guid jobId,
503      byte[] result,
504      double? percentage,
505      string exception,
506      bool finished) {
507
508      Logger.Info("BEGIN Job received for Storage - SUB method: " + jobId);
509
510      ResponseResultReceived response = new ResponseResultReceived();
511      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
512
513      SerializedJob job = new SerializedJob();
514
515      if (job != null) {
516        job.JobInfo = DaoLocator.JobDao.FindById(jobId);
517        if (job.JobInfo != null) {
518          job.JobInfo.Client = job.JobInfo.Client = DaoLocator.ClientDao.GetClientForJob(jobId);
519        }
520      }
521     
522      if (job != null && job.JobInfo == null) {
523        response.Success = false;
524        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
525        response.JobId = jobId;
526
527        Logger.Error("No job with Id " + jobId);
528
529        //tx.Rollback();
530        return response;
531      }
532      if (job.JobInfo.State == State.Abort) {
533        response.Success = false;
534        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_WAS_ABORTED;
535
536        Logger.Error("Job was aborted! " + job.JobInfo);
537
538        //tx.Rollback();
539        return response;
540      }
541      if (job.JobInfo.Client == null) {
542        response.Success = false;
543        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
544        response.JobId = jobId;
545
546        Logger.Error("Job is not being calculated (client = null)! " + job.JobInfo);
547
548        //tx.Rollback();
549        return response;
550      }
551      if (job.JobInfo.Client.Id != clientId) {
552        response.Success = false;
553        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
554        response.JobId = jobId;
555
556        Logger.Error("Wrong Client for this Job! " + job.JobInfo + ", Sending Client is: " + clientId);
557
558        //tx.Rollback();
559        return response;
560      }
561      if (job.JobInfo.State == State.Finished) {
562        response.Success = true;
563        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
564        response.JobId = jobId;
565
566        Logger.Error("Job already finished! " + job.JobInfo + ", Sending Client is: " + clientId);
567
568        //tx.Rollback();
569        return response;
570      }
571      //Todo: RequestsnapshotSent => calculating?
572      if (job.JobInfo.State == State.RequestSnapshotSent) {
573        job.JobInfo.State = State.Calculating;
574      }
575      if (job.JobInfo.State != State.Calculating &&
576        job.JobInfo.State != State.Pending) {
577        response.Success = false;
578        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
579        response.JobId = jobId;
580
581        Logger.Error("Wrong Job State, job is: " + job.JobInfo);
582
583        //tx.Rollback();
584        return response;
585      }
586      job.JobInfo.Percentage = percentage;
587
588      if (!string.IsNullOrEmpty(exception)) {
589        job.JobInfo.State = State.Failed;
590        job.JobInfo.Exception = exception;
591        job.JobInfo.DateFinished = DateTime.Now;
592      } else if (finished) {
593        job.JobInfo.State = State.Finished;
594        job.JobInfo.DateFinished = DateTime.Now;
595      }
596
597      job.SerializedJobData = result;
598
599      DaoLocator.JobDao.Update(job.JobInfo);
600
601      response.Success = true;
602      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
603      response.JobId = jobId;
604      response.Finished = finished;
605
606      Logger.Info("END Job received for Storage - SUB method: " + jobId);
607      return response;
608
609    }
610
611    /// <summary>
612    /// the client can send job results during calculating
613    /// and will send a final job result when he finished calculating
614    /// these job results will be stored in the database
615    /// </summary>
616    /// <param name="clientId"></param>
617    /// <param name="jobId"></param>
618    /// <param name="result"></param>
619    /// <param name="exception"></param>
620    /// <param name="finished"></param>
621    /// <returns></returns>
622    public ResponseResultReceived StoreFinishedJobResult(Guid clientId,
623      Guid jobId,
624      byte[] result,
625      double percentage,
626      string exception) {
627
628      return ProcessJobResult(clientId, jobId, result, percentage, exception, true);
629    }
630
631    public ResponseResultReceived ProcessSnapshot(Guid clientId, Guid jobId, byte[] result, double percentage, string exception) {
632      return ProcessJobResult(clientId, jobId, result, percentage, exception, false);
633    }
634
635    /// <summary>
636    /// when a client logs out the state will be set
637    /// and the entry in the last hearbeats dictionary will be removed
638    /// </summary>
639    /// <param name="clientId"></param>
640    /// <returns></returns>                       
641    public Response Logout(Guid clientId) {
642      Logger.Info("Client logged out " + clientId);
643
644      Response response = new Response();
645
646      heartbeatLock.EnterWriteLock();
647      if (lastHeartbeats.ContainsKey(clientId))
648        lastHeartbeats.Remove(clientId);
649      heartbeatLock.ExitWriteLock();
650
651      ClientDto client = DaoLocator.ClientDao.FindById(clientId);
652      if (client == null) {
653        response.Success = false;
654        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
655        return response;
656      }
657      if (client.State == State.Calculating) {
658        // check wich job the client was calculating and reset it
659        IEnumerable<JobDto> jobsOfClient = DaoLocator.JobDao.FindActiveJobsOfClient(client);
660        foreach (JobDto job in jobsOfClient) {
661          if (job.State != State.Finished)
662            DaoLocator.JobDao.SetJobOffline(job);
663        }
664      }
665
666      client.State = State.Offline;
667      DaoLocator.ClientDao.Update(client);
668
669      response.Success = true;
670      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
671
672      return response;
673    }
674
675    /// <summary>
676    /// If a client goes offline and restores a job he was calculating
677    /// he can ask the client if he still needs the job result
678    /// </summary>
679    /// <param name="jobId"></param>
680    /// <returns></returns>
681    public Response IsJobStillNeeded(Guid jobId) {
682      Response response = new Response();
683      JobDto job = DaoLocator.JobDao.FindById(jobId);
684      if (job == null) {
685        response.Success = false;
686        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_DOESNT_EXIST;
687        Logger.Error("Job doesn't exist (anymore)! " + jobId);
688        return response;
689      }
690      if (job.State == State.Finished) {
691        response.Success = true;
692        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_ALLREADY_FINISHED;
693        Logger.Error("already finished! " + job);
694        return response;
695      }
696      job.State = State.Pending;
697      lock (pendingJobs) {
698        pendingJobs.Add(job.Id, PENDING_TIMEOUT);
699      }
700
701      DaoLocator.JobDao.Update(job);
702
703      response.Success = true;
704      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_SEND_JOBRESULT;
705      return response;
706    }
707
708    public ResponsePlugin SendPlugins(List<HivePluginInfoDto> pluginList) {
709      ResponsePlugin response = new ResponsePlugin();
710      foreach (HivePluginInfoDto pluginInfo in pluginList) {
711        if (pluginInfo.Update) {
712          //check if there is a newer version         
713          IPluginDescription ipd =
714            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();
715          if (ipd != null) {
716            response.Plugins.Add(convertPluginDescriptorToDto(ipd));
717          }
718        } else {
719          IPluginDescription ipd =
720            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();
721          if (ipd != null) {
722            response.Plugins.Add(convertPluginDescriptorToDto(ipd));
723          } else {
724            response.Success = false;
725            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_PLUGINS_NOT_AVAIL;
726            return response;
727          }
728        }
729      }
730      response.Success = true;
731      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_PLUGINS_SENT;
732
733      return response;
734    }
735
736    private CachedHivePluginInfoDto convertPluginDescriptorToDto(IPluginDescription currPlugin) {
737      CachedHivePluginInfoDto currCachedPlugin = new CachedHivePluginInfoDto {
738        Name = currPlugin.Name,
739        Version = currPlugin.Version
740      };
741
742      foreach (string fileName in from file in currPlugin.Files select file.Name) {
743        currCachedPlugin.PluginFiles.Add(new HivePluginFile(File.ReadAllBytes(fileName), fileName));
744      }
745      return currCachedPlugin;
746    }
747
748    #endregion
749  }
750}
Note: See TracBrowser for help on using the repository browser.