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

Last change on this file since 1166 was 1166, checked in by svonolfe, 12 years ago

Improved memory consumption, fixed bug that already calculated jobs where reset (#372)

File size: 13.9 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        response.Job = job2Calculate;
239        jobAdapter.Update(job2Calculate);
240        response.Success = true;
241        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
242      } else {
243        response.Success = true;
244        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
245      }
246
247      jobLock.ReleaseMutex();
248      /// End Critical section ///
249
250      return response;
251    }
252
253    /// <summary>
254    /// the client can send job results during calculating
255    /// and will send a final job result when he finished calculating
256    /// these job results will be stored in the database
257    /// </summary>
258    /// <param name="clientId"></param>
259    /// <param name="jobId"></param>
260    /// <param name="result"></param>
261    /// <param name="exception"></param>
262    /// <param name="finished"></param>
263    /// <returns></returns>
264    public ResponseResultReceived SendJobResult(Guid clientId,
265      long jobId,
266      byte[] result,
267      double percentage,
268      Exception exception, 
269      bool finished) {
270      ResponseResultReceived response = new ResponseResultReceived();
271      ClientInfo client =
272        clientAdapter.GetById(clientId);
273
274      Job job =
275        jobAdapter.GetById(jobId);
276
277      if (job.Client == null)    {
278        response.Success = false;
279        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
280        return response;
281      }
282      if (job.Client.ClientId != clientId) {
283        response.Success = false;
284        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
285        return response;
286      }
287      if (job == null) {
288        response.Success = false;
289        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
290        return response;
291      }
292      if (job.State != State.calculating) {
293        response.Success = false;
294        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
295        return response;
296      }
297      job.SerializedJob = result;
298      job.Percentage = percentage;
299
300      if (finished) {
301        job.State = State.finished;
302        jobAdapter.Update(job);
303
304        client.State = State.idle;
305        clientAdapter.Update(client);
306
307        List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
308        foreach (JobResult currentResult in jobResults)
309          jobResultAdapter.Delete(currentResult);
310      }
311
312      JobResult jobResult =
313        new JobResult();
314      jobResult.Client = client;
315      jobResult.Job = job;
316      jobResult.Result = result;
317      jobResult.Percentage = percentage;
318      jobResult.Exception = exception;
319
320      jobResultAdapter.Update(jobResult);
321      jobAdapter.Update(job);
322
323      response.Success = true;
324      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
325      response.JobId = jobId;
326      response.finished = finished;
327
328      return response;
329    }
330
331    /// <summary>
332    /// when a client logs out the state will be set
333    /// and the entry in the last hearbeats dictionary will be removed
334    /// </summary>
335    /// <param name="clientId"></param>
336    /// <returns></returns>                       
337    public Response Logout(Guid clientId) {
338      Response response = new Response();
339
340      heartbeatLock.EnterWriteLock();
341      if (lastHeartbeats.ContainsKey(clientId))
342        lastHeartbeats.Remove(clientId);
343      heartbeatLock.ExitWriteLock();
344
345      ClientInfo client = clientAdapter.GetById(clientId);
346      if (client == null) {
347        response.Success = false;
348        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
349        return response;
350      }
351      List<Job> allJobs = new List<Job>(jobAdapter.GetAll());
352      if (client.State == State.calculating) {
353        // check wich job the client was calculating and reset it
354        foreach (Job job in allJobs) {
355          if (job.Client.ClientId == client.ClientId) {
356            jobManager.ResetJobsDependingOnResults(job);
357          }
358        }
359      }
360
361      client.State = State.offline;
362      clientAdapter.Update(client);
363
364      response.Success = true;
365      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
366     
367      return response;
368    }
369
370    #endregion
371  }
372}
Note: See TracBrowser for help on using the repository browser.