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

Last change on this file since 1170 was 1170, checked in by msteinbi, 12 years ago

Implementing Lifecycle Management (#453)

File size: 14.0 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.Core.InternalInterfaces.DataAccess;
31using System.Resources;
32using System.Reflection;
33using HeuristicLab.Hive.JobBase;
34using HeuristicLab.Hive.Server.Core.InternalInterfaces;
35using System.Threading;
36
37namespace HeuristicLab.Hive.Server.Core {
38  /// <summary>
39  /// The ClientCommunicator manages the whole communication with the client
40  /// </summary>
41  public class ClientCommunicator: IClientCommunicator {
42    private static Dictionary<Guid, DateTime> lastHeartbeats =
43      new Dictionary<Guid,DateTime>();
44
45    private static ReaderWriterLockSlim heartbeatLock =
46      new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
47
48    private static Mutex jobLock =
49      new Mutex();
50
51    IClientAdapter clientAdapter;
52    IJobAdapter jobAdapter;
53    IJobResultsAdapter jobResultAdapter;
54    ILifecycleManager lifecycleManager;
55    IInternalJobManager jobManager;
56
57    /// <summary>
58    /// Initialization of the Adapters to the database
59    /// Initialization of Eventhandler for the lifecycle management
60    /// Initialization of lastHearbeats Dictionary
61    /// </summary>
62    public ClientCommunicator() {
63      clientAdapter = ServiceLocator.GetClientAdapter();
64      jobAdapter = ServiceLocator.GetJobAdapter();
65      jobResultAdapter = ServiceLocator.GetJobResultsAdapter();
66      lifecycleManager = ServiceLocator.GetLifecycleManager();
67      jobManager = ServiceLocator.GetJobManager() as
68        IInternalJobManager;
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.ClientId)) {
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.ClientId];
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.ClientId);
113              heartbeatLock.ExitWriteLock();
114            }
115          }
116
117          heartbeatLock.ExitUpgradeableReadLock();
118        } else {
119          heartbeatLock.EnterWriteLock();
120          if (lastHeartbeats.ContainsKey(client.ClientId))
121            lastHeartbeats.Remove(client.ClientId);
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.ClientId)) {
140        lastHeartbeats[clientInfo.ClientId] = DateTime.Now;
141      } else {
142        lastHeartbeats.Add(clientInfo.ClientId, DateTime.Now);
143      }
144      heartbeatLock.ExitWriteLock();
145
146      ICollection<ClientInfo> allClients = clientAdapter.GetAll();
147      ClientInfo client = clientAdapter.GetById(clientInfo.ClientId);
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 SendHeartBeat(HeartBeatData hbData) {
169      ResponseHB response = new ResponseHB();
170
171      response.ActionRequest = new List<MessageContainer>();
172      if (clientAdapter.GetById(hbData.ClientId).State == State.offline ||
173          clientAdapter.GetById(hbData.ClientId).State == State.nullState) {
174        response.Success = false;
175        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_USER_NOT_LOGGED_IN;
176        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
177        return response;
178      }
179
180      heartbeatLock.EnterWriteLock();
181      if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
182        lastHeartbeats[hbData.ClientId] = DateTime.Now;
183      } else {
184        lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
185      }
186      heartbeatLock.ExitWriteLock();
187
188      response.Success = true;
189      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_HARDBEAT_RECEIVED;
190      List<Job> allOfflineJobs = new List<Job>(jobAdapter.GetJobsByState(State.offline));
191      if (allOfflineJobs.Count > 0 && hbData.freeCores > 0)
192        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
193      else
194        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
195
196      if (hbData.jobProgress != null) {
197        List<Job> jobsOfClient = new List<Job>(jobAdapter.GetActiveJobsOf(clientAdapter.GetById(hbData.ClientId)));
198        if (jobsOfClient == null || jobsOfClient.Count == 0) {
199          response.Success = false;
200          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
201          return response;
202        }
203
204        foreach (KeyValuePair<long, double> jobProgress in hbData.jobProgress) {
205          Job curJob = jobAdapter.GetById(jobProgress.Key);
206          if (curJob.Client == null || curJob.Client.ClientId != hbData.ClientId) {
207            response.Success = false;
208            response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
209          } else {
210            curJob.Percentage = jobProgress.Value;
211            jobAdapter.Update(curJob);
212          }
213        }
214      }
215
216      return response;
217    }
218   
219    /// <summary>
220    /// if the client asked to pull a job he calls this method
221    /// the server selects a job and sends it to the client
222    /// </summary>
223    /// <param name="clientId"></param>
224    /// <returns></returns>
225    public ResponseJob PullJob(Guid clientId) {
226      ResponseJob response = new ResponseJob();
227     
228      /// Critical section ///
229      jobLock.WaitOne();
230
231      LinkedList<Job> allOfflineJobs = new LinkedList<Job>(jobAdapter.GetJobsByState(State.offline));
232      if (allOfflineJobs != null && allOfflineJobs.Count > 0) {
233        Job job2Calculate = allOfflineJobs.First.Value;
234        job2Calculate.State = State.calculating;
235        job2Calculate.Client = clientAdapter.GetById(clientId);
236        job2Calculate.Client.State = State.calculating;
237
238        job2Calculate.DateCalculated = DateTime.Now;
239        response.Job = job2Calculate;
240        jobAdapter.Update(job2Calculate);
241        response.Success = true;
242        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
243      } else {
244        response.Success = true;
245        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
246      }
247
248      jobLock.ReleaseMutex();
249      /// End Critical section ///
250
251      return response;
252    }
253
254    /// <summary>
255    /// the client can send job results during calculating
256    /// and will send a final job result when he finished calculating
257    /// these job results will be stored in the database
258    /// </summary>
259    /// <param name="clientId"></param>
260    /// <param name="jobId"></param>
261    /// <param name="result"></param>
262    /// <param name="exception"></param>
263    /// <param name="finished"></param>
264    /// <returns></returns>
265    public ResponseResultReceived SendJobResult(Guid clientId,
266      long jobId,
267      byte[] result,
268      double percentage,
269      Exception exception, 
270      bool finished) {
271      ResponseResultReceived response = new ResponseResultReceived();
272      ClientInfo client =
273        clientAdapter.GetById(clientId);
274
275      Job job =
276        jobAdapter.GetById(jobId);
277
278      if (job.Client == null)    {
279        response.Success = false;
280        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
281        return response;
282      }
283      if (job.Client.ClientId != clientId) {
284        response.Success = false;
285        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
286        return response;
287      }
288      if (job == null) {
289        response.Success = false;
290        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
291        return response;
292      }
293      if (job.State != State.calculating) {
294        response.Success = false;
295        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
296        return response;
297      }
298      job.SerializedJob = result;
299      job.Percentage = percentage;
300
301      if (finished) {
302        job.State = State.finished;
303        jobAdapter.Update(job);
304
305        client.State = State.idle;
306        clientAdapter.Update(client);
307
308        List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
309        foreach (JobResult currentResult in jobResults)
310          jobResultAdapter.Delete(currentResult);
311      }
312
313      JobResult jobResult =
314        new JobResult();
315      jobResult.Client = client;
316      jobResult.Job = job;
317      jobResult.Result = result;
318      jobResult.Percentage = percentage;
319      jobResult.Exception = exception;
320      jobResult.DateFinished = DateTime.Now;
321
322      jobResultAdapter.Update(jobResult);
323      jobAdapter.Update(job);
324
325      response.Success = true;
326      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
327      response.JobId = jobId;
328      response.finished = finished;
329
330      return response;
331    }
332
333    /// <summary>
334    /// when a client logs out the state will be set
335    /// and the entry in the last hearbeats dictionary will be removed
336    /// </summary>
337    /// <param name="clientId"></param>
338    /// <returns></returns>                       
339    public Response Logout(Guid clientId) {
340      Response response = new Response();
341
342      heartbeatLock.EnterWriteLock();
343      if (lastHeartbeats.ContainsKey(clientId))
344        lastHeartbeats.Remove(clientId);
345      heartbeatLock.ExitWriteLock();
346
347      ClientInfo client = clientAdapter.GetById(clientId);
348      if (client == null) {
349        response.Success = false;
350        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
351        return response;
352      }
353      List<Job> allJobs = new List<Job>(jobAdapter.GetAll());
354      if (client.State == State.calculating) {
355        // check wich job the client was calculating and reset it
356        foreach (Job job in allJobs) {
357          if (job.Client.ClientId == client.ClientId) {
358            jobManager.ResetJobsDependingOnResults(job);
359          }
360        }
361      }
362
363      client.State = State.offline;
364      clientAdapter.Update(client);
365
366      response.Success = true;
367      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
368     
369      return response;
370    }
371
372    #endregion
373  }
374}
Note: See TracBrowser for help on using the repository browser.