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

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

Implementing Lifecycle Management (#453)

File size: 13.1 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 System.Runtime.CompilerServices;
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    Dictionary<Guid, DateTime> lastHeartbeats =
42      new Dictionary<Guid,DateTime>();
43
44    IClientAdapter clientAdapter;
45    IJobAdapter jobAdapter;
46    IJobResultsAdapter jobResultAdapter;
47    ILifecycleManager lifecycleManager;
48
49    /// <summary>
50    /// Initialization of the Adapters to the database
51    /// Initialization of Eventhandler for the lifecycle management
52    /// Initialization of lastHearbeats Dictionary
53    /// </summary>
54    public ClientCommunicator() {
55      clientAdapter = ServiceLocator.GetClientAdapter();
56      jobAdapter = ServiceLocator.GetJobAdapter();
57      jobResultAdapter = ServiceLocator.GetJobResultsAdapter();
58      lifecycleManager = ServiceLocator.GetLifecycleManager();
59
60      lifecycleManager.RegisterHeartbeat(
61        new EventHandler(lifecycleManager_OnServerHeartbeat));
62
63      lastHeartbeats = new Dictionary<Guid, DateTime>();
64    }
65
66    /// <summary>
67    /// Check if online clients send their hearbeats
68    /// if not -> set them offline and check if they where calculating a job
69    /// </summary>
70    /// <param name="sender"></param>
71    /// <param name="e"></param>
72    [MethodImpl(MethodImplOptions.Synchronized)]
73    void lifecycleManager_OnServerHeartbeat(object sender, EventArgs e) {
74      List<ClientInfo> allClients = new List<ClientInfo>(clientAdapter.GetAll());
75      List<Job> allJobs = new List<Job>(jobAdapter.GetAll());
76
77      foreach (ClientInfo client in allClients) {
78        if (client.State != State.offline && client.State != State.nullState) {
79          if (!lastHeartbeats.ContainsKey(client.ClientId)) {
80            client.State = State.offline;
81            clientAdapter.Update(client);
82          } else {
83            DateTime lastHbOfClient = lastHeartbeats[client.ClientId];
84            TimeSpan dif = DateTime.Now.Subtract(lastHbOfClient);
85            // check if time between last hearbeat and now is greather than HEARTBEAT_MAX_DIF
86            if (dif.Seconds > ApplicationConstants.HEARTBEAT_MAX_DIF) {
87              // if client calculated jobs, the job must be reset
88              if (client.State == State.calculating) {
89                // check wich job the client was calculating and reset it
90                foreach (Job job in allJobs) {
91                  if (job.Client.ClientId == client.ClientId) {
92                    List<JobResult> allJobResults = new List<JobResult>(jobResultAdapter.GetAll());
93                    foreach (JobResult jR in allJobResults) {
94                      JobResult lastJobResult = null;
95                      if (jR.Job != null && jR.Job.Id == job.Id) {
96                        if (lastJobResult != null) {
97
98                        }
99                      }
100                    }
101
102
103                    job.Client = null;
104                    job.Percentage = 0;
105                    job.State = State.idle;
106                  }
107                }
108              }
109             
110              // client must be set offline
111              client.State = State.offline;
112              clientAdapter.Update(client);
113              lastHeartbeats.Remove(client.ClientId);
114            }
115          }
116        } else {
117          if (lastHeartbeats.ContainsKey(client.ClientId))
118            lastHeartbeats.Remove(client.ClientId);
119        }
120      }
121    }
122
123    #region IClientCommunicator Members
124
125    /// <summary>
126    /// Login process for the client
127    /// A hearbeat entry is created as well (login is the first hearbeat)
128    /// </summary>
129    /// <param name="clientInfo"></param>
130    /// <returns></returns>
131    [MethodImpl(MethodImplOptions.Synchronized)]
132    public Response Login(ClientInfo clientInfo) {
133      Response response = new Response();
134
135      if (lastHeartbeats.ContainsKey(clientInfo.ClientId)) {
136        lastHeartbeats[clientInfo.ClientId] = DateTime.Now;
137      } else {
138        lastHeartbeats.Add(clientInfo.ClientId, DateTime.Now);
139      }
140
141      ICollection<ClientInfo> allClients = clientAdapter.GetAll();
142      ClientInfo client = clientAdapter.GetById(clientInfo.ClientId);
143      if (client != null && client.State != State.offline && client.State != State.nullState) {
144        response.Success = false;
145        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_USER_ALLREADY_ONLINE;
146        return response;
147      }
148      clientInfo.State = State.idle;
149      clientAdapter.Update(clientInfo);
150      response.Success = true;
151      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_SUCCESS;
152
153      return response;
154    }
155
156    /// <summary>
157    /// The client has to send regulary heartbeats
158    /// this hearbeats will be stored in the heartbeats dictionary
159    /// check if there is work for the client and send the client a response if he should pull a job
160    /// </summary>
161    /// <param name="hbData"></param>
162    /// <returns></returns>
163    [MethodImpl(MethodImplOptions.Synchronized)]
164    public ResponseHB SendHeartBeat(HeartBeatData hbData) {
165      ResponseHB response = new ResponseHB();
166
167      response.ActionRequest = new List<MessageContainer>();
168      if (clientAdapter.GetById(hbData.ClientId).State == State.offline ||
169          clientAdapter.GetById(hbData.ClientId).State == State.nullState) {
170        response.Success = false;
171        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_USER_NOT_LOGGED_IN;
172        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
173        return response;
174      }
175
176      if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
177        lastHeartbeats[hbData.ClientId] = DateTime.Now;
178      } else {
179        lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
180      }
181
182      response.Success = true;
183      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_HARDBEAT_RECEIVED;
184      List<Job> allOfflineJobs = new List<Job>(jobAdapter.GetJobsByState(State.offline));
185      if (allOfflineJobs.Count > 0 && hbData.freeCores > 0)
186        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
187      else
188        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
189
190      if (hbData.jobProgress != null) {
191        foreach (KeyValuePair<long, double> jobProgress in hbData.jobProgress) {
192          Job curJob = jobAdapter.GetById(jobProgress.Key);
193          curJob.Percentage = jobProgress.Value;
194          jobAdapter.Update(curJob);
195        }
196      }
197
198      return response;
199    }
200   
201    /// <summary>
202    /// if the client asked to pull a job he calls this method
203    /// the server selects a job and sends it to the client
204    /// </summary>
205    /// <param name="clientId"></param>
206    /// <returns></returns>
207    [MethodImpl(MethodImplOptions.Synchronized)]
208    public ResponseJob PullJob(Guid clientId) {
209      ResponseJob response = new ResponseJob();
210      lock (this) {
211        LinkedList<Job> allOfflineJobs = new LinkedList<Job>(jobAdapter.GetJobsByState(State.offline));
212        if (allOfflineJobs != null && allOfflineJobs.Count > 0) {
213          Job job2Calculate = allOfflineJobs.First.Value;
214          job2Calculate.State = State.calculating;
215          job2Calculate.Client = clientAdapter.GetById(clientId);
216          response.Job = job2Calculate;
217          jobAdapter.Update(job2Calculate);
218          response.Success = true;
219          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
220          return response;
221        }
222      }
223      response.Success = true;
224      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
225      return response;
226    }
227
228    /// <summary>
229    /// the client can send job results during calculating
230    /// and will send a final job result when he finished calculating
231    /// these job results will be stored in the database
232    /// </summary>
233    /// <param name="clientId"></param>
234    /// <param name="jobId"></param>
235    /// <param name="result"></param>
236    /// <param name="exception"></param>
237    /// <param name="finished"></param>
238    /// <returns></returns>
239    [MethodImpl(MethodImplOptions.Synchronized)]
240    public ResponseResultReceived SendJobResult(Guid clientId,
241      long jobId,
242      byte[] result,
243      double percentage,
244      Exception exception, 
245      bool finished) {
246      ResponseResultReceived response = new ResponseResultReceived();
247      ClientInfo client =
248        clientAdapter.GetById(clientId);
249
250      Job job =
251        jobAdapter.GetById(jobId);
252
253      if (job.Client == null)    {
254        response.Success = false;
255        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_IS_NOT_BEEING_CALCULATED;
256        return response;
257      }
258      if (job.Client.ClientId != clientId) {
259        response.Success = false;
260        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_CLIENT_FOR_JOB;
261        return response;
262      }
263      if (job == null) {
264        response.Success = false;
265        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOB_WITH_THIS_ID;
266        return response;
267      }
268      if (job.State != State.calculating) {
269        response.Success = false;
270        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
271        return response;
272      }
273      job.SerializedJob = result;
274      job.Percentage = percentage;
275
276      if (finished) {
277        job.State = State.finished;
278        jobAdapter.Update(job);
279
280        List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
281        foreach (JobResult currentResult in jobResults)
282          jobResultAdapter.Delete(currentResult);
283      }
284
285      JobResult jobResult =
286        new JobResult();
287      jobResult.Client = client;
288      jobResult.Job = job;
289      jobResult.Result = result;
290      jobResult.Percentage = percentage;
291      jobResult.Exception = exception;
292
293      jobResultAdapter.Update(jobResult);
294      jobAdapter.Update(job);
295
296      response.Success = true;
297      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
298      response.JobId = jobId;
299
300      return response;
301    }
302
303    /// <summary>
304    /// when a client logs out the state will be set
305    /// and the entry in the last hearbeats dictionary will be removed
306    /// </summary>
307    /// <param name="clientId"></param>
308    /// <returns></returns>
309    [MethodImpl(MethodImplOptions.Synchronized)]                       
310    public Response Logout(Guid clientId) {
311      Response response = new Response();
312
313      if (lastHeartbeats.ContainsKey(clientId))
314        lastHeartbeats.Remove(clientId);
315
316      ClientInfo client = clientAdapter.GetById(clientId);
317      if (client == null) {
318        response.Success = false;
319        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
320        return response;
321      }
322      List<Job> allJobs = new List<Job>(jobAdapter.GetAll());
323      if (client.State == State.calculating) {
324        // check wich job the client was calculating and reset it
325        foreach (Job job in allJobs) {
326          if (job.Client.ClientId == client.ClientId) {
327            // TODO check for job results
328            job.Client = null;
329            job.Percentage = 0;
330            job.State = State.idle;
331          }
332        }
333      }
334
335      client.State = State.offline;
336      clientAdapter.Update(client);
337
338      response.Success = true;
339      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
340     
341      return response;
342    }
343
344    #endregion
345  }
346}
Note: See TracBrowser for help on using the repository browser.