Free cookie consent management tool by TermsFeed Policy Generator

source: trunk/sources/HeuristicLab.Hive.Server.Core/ClientCommunicator.cs @ 1449

Last change on this file since 1449 was 1449, checked in by svonolfe, 15 years ago

Refactored DAL (now using GUIDs as IDs instead of longs) (#527)

File size: 15.8 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.Linq;
25using System.Text;
26using HeuristicLab.Hive.Contracts.BusinessObjects;
27using HeuristicLab.Hive.Contracts.Interfaces;
28using HeuristicLab.Hive.Contracts;
29using HeuristicLab.Core;
30using HeuristicLab.Hive.Server.DataAccess;
31using System.Resources;
32using System.Reflection;
33using HeuristicLab.Hive.JobBase;
34using HeuristicLab.Hive.Server.Core.InternalInterfaces;
35using System.Threading;
36using HeuristicLab.PluginInfrastructure;
37
38namespace HeuristicLab.Hive.Server.Core {
39  /// <summary>
40  /// The ClientCommunicator manages the whole communication with the client
41  /// </summary>
42  public class ClientCommunicator: IClientCommunicator {
43    private static Dictionary<Guid, DateTime> lastHeartbeats =
44      new Dictionary<Guid,DateTime>();
45
46    private static ReaderWriterLockSlim heartbeatLock =
47      new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
48
49    private IClientAdapter clientAdapter;
50    private IJobAdapter jobAdapter;
51    private IJobResultsAdapter jobResultAdapter;
52    private ILifecycleManager lifecycleManager;
53    private IInternalJobManager jobManager;
54    private IScheduler scheduler;
55
56    /// <summary>
57    /// Initialization of the Adapters to the database
58    /// Initialization of Eventhandler for the lifecycle management
59    /// Initialization of lastHearbeats Dictionary
60    /// </summary>
61    public ClientCommunicator() {
62      clientAdapter = ServiceLocator.GetClientAdapter();
63      jobAdapter = ServiceLocator.GetJobAdapter();
64      jobResultAdapter = ServiceLocator.GetJobResultsAdapter();
65      lifecycleManager = ServiceLocator.GetLifecycleManager();
66      jobManager = ServiceLocator.GetJobManager() as
67        IInternalJobManager;
68      scheduler = ServiceLocator.GetScheduler();
69
70      lifecycleManager.RegisterHeartbeat(
71        new EventHandler(lifecycleManager_OnServerHeartbeat));
72    }
73
74    /// <summary>
75    /// Check if online clients send their hearbeats
76    /// if not -> set them offline and check if they where calculating a job
77    /// </summary>
78    /// <param name="sender"></param>
79    /// <param name="e"></param>
80    void lifecycleManager_OnServerHeartbeat(object sender, EventArgs e) {
81      List<ClientInfo> allClients = new List<ClientInfo>(clientAdapter.GetAll());
82
83      foreach (ClientInfo client in allClients) {
84        if (client.State != State.offline && client.State != State.nullState) {
85          heartbeatLock.EnterUpgradeableReadLock();
86
87          if (!lastHeartbeats.ContainsKey(client.Id)) {
88            client.State = State.offline;
89            clientAdapter.Update(client);
90            foreach (Job job in jobAdapter.GetActiveJobsOf(client)) {
91              jobManager.ResetJobsDependingOnResults(job);
92            }
93          } else {
94            DateTime lastHbOfClient = lastHeartbeats[client.Id];
95
96            TimeSpan dif = DateTime.Now.Subtract(lastHbOfClient);
97            // check if time between last hearbeat and now is greather than HEARTBEAT_MAX_DIF
98            if (dif.Seconds > ApplicationConstants.HEARTBEAT_MAX_DIF) {
99              // if client calculated jobs, the job must be reset
100              if (client.State == State.calculating) {
101                // check wich job the client was calculating and reset it
102                foreach (Job job in jobAdapter.GetActiveJobsOf(client)) {
103                  jobManager.ResetJobsDependingOnResults(job);
104                }
105              }
106             
107              // client must be set offline
108              client.State = State.offline;
109              clientAdapter.Update(client);
110
111              heartbeatLock.EnterWriteLock();
112              lastHeartbeats.Remove(client.Id);
113              heartbeatLock.ExitWriteLock();
114            }
115          }
116
117          heartbeatLock.ExitUpgradeableReadLock();
118        } else {
119          heartbeatLock.EnterWriteLock();
120          if (lastHeartbeats.ContainsKey(client.Id))
121            lastHeartbeats.Remove(client.Id);
122          heartbeatLock.ExitWriteLock();
123        }
124      }
125    }
126
127    #region IClientCommunicator Members
128
129    /// <summary>
130    /// Login process for the client
131    /// A hearbeat entry is created as well (login is the first hearbeat)
132    /// </summary>
133    /// <param name="clientInfo"></param>
134    /// <returns></returns>
135    public Response Login(ClientInfo clientInfo) {
136      Response response = new Response();
137
138      heartbeatLock.EnterWriteLock();
139      if (lastHeartbeats.ContainsKey(clientInfo.Id)) {
140        lastHeartbeats[clientInfo.Id] = DateTime.Now;
141      } else {
142        lastHeartbeats.Add(clientInfo.Id, DateTime.Now);
143      }
144      heartbeatLock.ExitWriteLock();
145
146      // todo: allClients legacy ?
147      ClientInfo client = clientAdapter.GetById(clientInfo.Id);
148      if (client != null && client.State != State.offline && client.State != State.nullState) {
149        response.Success = false;
150        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_USER_ALLREADY_ONLINE;
151        return response;
152      }
153      clientInfo.State = State.idle;
154      clientAdapter.Update(clientInfo);
155      response.Success = true;
156      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_SUCCESS;
157
158      return response;
159    }
160
161    /// <summary>
162    /// The client has to send regulary heartbeats
163    /// this hearbeats will be stored in the heartbeats dictionary
164    /// check if there is work for the client and send the client a response if he should pull a job
165    /// </summary>
166    /// <param name="hbData"></param>
167    /// <returns></returns>
168    public ResponseHB ProcessHeartBeat(HeartBeatData hbData) {
169      ResponseHB response = new ResponseHB();
170
171      // check if the client is logged in
172      response.ActionRequest = new List<MessageContainer>();
173      if (clientAdapter.GetById(hbData.ClientId).State == State.offline ||
174          clientAdapter.GetById(hbData.ClientId).State == State.nullState) {
175        response.Success = false;
176        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_USER_NOT_LOGGED_IN;
177        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
178        return response;
179      }
180
181      // save timestamp of this heartbeat
182      heartbeatLock.EnterWriteLock();
183      if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
184        lastHeartbeats[hbData.ClientId] = DateTime.Now;
185      } else {
186        lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
187      }
188      heartbeatLock.ExitWriteLock();
189
190      response.Success = true;
191      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_HEARTBEAT_RECEIVED;
192      // check if client has a free core for a new job
193      // if true, ask scheduler for a new job for this client
194      if (hbData.FreeCores > 0 && scheduler.ExistsJobForClient(hbData))
195        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
196      else
197        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
198
199      if (hbData.JobProgress != null) {
200        List<Job> jobsOfClient = new List<Job>(jobAdapter.GetActiveJobsOf(clientAdapter.GetById(hbData.ClientId)));
201        if (jobsOfClient == null || jobsOfClient.Count == 0) {
202          response.Success = false;
203          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
204          return response;
205        }
206
207        foreach (KeyValuePair<Guid, double> jobProgress in hbData.JobProgress) {
208          Job curJob = jobAdapter.GetById(jobProgress.Key);
209          if (curJob.Client == null || curJob.Client.Id != hbData.ClientId) {
210            response.Success = false;
211            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
212          } else if(curJob.State == State.finished) {
213            // another client has finished this job allready
214            // the client can abort it
215            response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));       
216          } else {
217            // save job progress
218            curJob.Percentage = jobProgress.Value;
219            jobAdapter.Update(curJob);
220          }
221        }
222      }
223
224      return response;
225    }
226   
227    /// <summary>
228    /// if the client asked to pull a job he calls this method
229    /// the server selects a job and sends it to the client
230    /// </summary>
231    /// <param name="clientId"></param>
232    /// <returns></returns>
233    public ResponseJob SendJob(Guid clientId) {
234      ResponseJob response = new ResponseJob();
235
236      Job job2Calculate = scheduler.GetNextJobForClient(clientId);
237      if (job2Calculate != null) {
238        response.Job = job2Calculate;
239        response.Success = true;
240        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
241      } else {
242        response.Success = false;
243        response.Job = null;
244        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
245      }
246      return response;
247    }
248
249    private ResponseResultReceived ProcessJobResult(Guid clientId,
250      Guid jobId,
251      byte[] result,
252      double percentage,
253      Exception exception,
254      bool finished) {
255
256      ResponseResultReceived response = new ResponseResultReceived();
257      ClientInfo client =
258        clientAdapter.GetById(clientId);
259
260      Job job =
261        jobAdapter.GetById(jobId);
262
263      if (job.Client == null) {
264        response.Success = false;
265        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
266        return response;
267      }
268      if (job.Client.Id != clientId) {
269        response.Success = false;
270        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
271        return response;
272      }
273      if (job == null) {
274        response.Success = false;
275        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
276        return response;
277      }
278      if (job.State == State.finished) {
279        response.Success = true;
280        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
281        return response;
282      }
283      if (job.State != State.calculating) {
284        response.Success = false;
285        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
286        return response;
287      }
288      job.SerializedJob = result;
289      job.Percentage = percentage;
290
291      if (finished) {
292        job.State = State.finished;
293        jobAdapter.Update(job);
294
295        client.State = State.idle;
296        clientAdapter.Update(client);
297
298        List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
299        foreach (JobResult currentResult in jobResults)
300          jobResultAdapter.Delete(currentResult);
301      }
302
303      JobResult jobResult =
304        new JobResult();
305      jobResult.Client = client;
306      jobResult.Job = job;
307      jobResult.Result = result;
308      jobResult.Percentage = percentage;
309      jobResult.Exception = exception;
310      jobResult.DateFinished = DateTime.Now;
311
312      jobResultAdapter.Update(jobResult);
313      jobAdapter.Update(job);
314
315      response.Success = true;
316      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
317      response.JobId = jobId;
318      response.finished = finished;
319
320      return response;
321    }
322
323
324    /// <summary>
325    /// the client can send job results during calculating
326    /// and will send a final job result when he finished calculating
327    /// these job results will be stored in the database
328    /// </summary>
329    /// <param name="clientId"></param>
330    /// <param name="jobId"></param>
331    /// <param name="result"></param>
332    /// <param name="exception"></param>
333    /// <param name="finished"></param>
334    /// <returns></returns>
335    public ResponseResultReceived StoreFinishedJobResult(Guid clientId,
336      Guid jobId,
337      byte[] result,
338      double percentage,
339      Exception exception) {
340
341      return ProcessJobResult(clientId, jobId, result, percentage, exception, true);
342    }
343
344
345    public ResponseResultReceived ProcessSnapshot(Guid clientId, Guid jobId, byte[] result, double percentage, Exception exception) {
346      return ProcessJobResult(clientId, jobId, result, percentage, exception, false);
347    }
348
349    /// <summary>
350    /// when a client logs out the state will be set
351    /// and the entry in the last hearbeats dictionary will be removed
352    /// </summary>
353    /// <param name="clientId"></param>
354    /// <returns></returns>                       
355    public Response Logout(Guid clientId) {
356      Response response = new Response();
357
358      heartbeatLock.EnterWriteLock();
359      if (lastHeartbeats.ContainsKey(clientId))
360        lastHeartbeats.Remove(clientId);
361      heartbeatLock.ExitWriteLock();
362
363      ClientInfo client = clientAdapter.GetById(clientId);
364      if (client == null) {
365        response.Success = false;
366        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
367        return response;
368      }
369      List<Job> allJobs = new List<Job>(jobAdapter.GetAll());
370      if (client.State == State.calculating) {
371        // check wich job the client was calculating and reset it
372        foreach (Job job in allJobs) {
373          if (job.Client.Id == client.Id) {
374            jobManager.ResetJobsDependingOnResults(job);
375          }
376        }
377      }
378
379      client.State = State.offline;
380      clientAdapter.Update(client);
381
382      response.Success = true;
383      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
384     
385      return response;
386    }
387
388    /// <summary>
389    /// If a client goes offline and restores a job he was calculating
390    /// he can ask the client if he still needs the job result
391    /// </summary>
392    /// <param name="jobId"></param>
393    /// <returns></returns>
394    public Response IsJobStillNeeded(Guid jobId) {
395      Response response = new Response();
396      Job job = jobAdapter.GetById(jobId);
397      if (job == null) {
398        response.Success = false;
399        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_DOESNT_EXIST;
400        return response;
401      }
402      if (job.State == State.finished) {
403        response.Success = true;
404        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_ALLREADY_FINISHED;
405        return response;
406      }
407      job.State = State.finished;
408      jobAdapter.Update(job);
409     
410      response.Success = true;
411      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_SEND_JOBRESULT;
412      return response;
413    }
414
415    public ResponsePlugin SendPlugins(List<PluginInfo> pluginList) {
416      throw new NotImplementedException();
417    }
418
419    #endregion
420  }
421}
Note: See TracBrowser for help on using the repository browser.