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

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

Refactoring of Job progress (#466)

File size: 11.6 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.OnServerHeartbeat +=
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                    // TODO check for job results
93                    job.Client = null;
94                    job.Percentage = 0;
95                    job.State = State.idle;
96                  }
97                }
98              }
99             
100              // client must be set offline
101              client.State = State.offline;
102              clientAdapter.Update(client);
103              lastHeartbeats.Remove(client.ClientId);
104            }
105          }
106        } else {
107          if (lastHeartbeats.ContainsKey(client.ClientId))
108            lastHeartbeats.Remove(client.ClientId);
109        }
110      }
111    }
112
113    #region IClientCommunicator Members
114
115    /// <summary>
116    /// Login process for the client
117    /// A hearbeat entry is created as well (login is the first hearbeat)
118    /// </summary>
119    /// <param name="clientInfo"></param>
120    /// <returns></returns>
121    [MethodImpl(MethodImplOptions.Synchronized)]
122    public Response Login(ClientInfo clientInfo) {
123      Response response = new Response();
124
125      if (lastHeartbeats.ContainsKey(clientInfo.ClientId)) {
126        lastHeartbeats[clientInfo.ClientId] = DateTime.Now;
127      } else {
128        lastHeartbeats.Add(clientInfo.ClientId, DateTime.Now);
129      }
130
131      ICollection<ClientInfo> allClients = clientAdapter.GetAll();
132      ClientInfo client = clientAdapter.GetById(clientInfo.ClientId);
133      if (client != null && client.State != State.offline && client.State != State.nullState) {
134        response.Success = false;
135        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_USER_ALLREADY_ONLINE;
136        return response;
137      }
138      clientInfo.State = State.idle;
139      clientAdapter.Update(clientInfo);
140      response.Success = true;
141      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGIN_SUCCESS;
142
143      return response;
144    }
145
146    /// <summary>
147    /// The client has to send regulary heartbeats
148    /// this hearbeats will be stored in the heartbeats dictionary
149    /// check if there is work for the client and send the client a response if he should pull a job
150    /// </summary>
151    /// <param name="hbData"></param>
152    /// <returns></returns>
153    [MethodImpl(MethodImplOptions.Synchronized)]
154    public ResponseHB SendHeartBeat(HeartBeatData hbData) {
155      ResponseHB response = new ResponseHB();
156
157      response.ActionRequest = new List<MessageContainer>();
158      if (clientAdapter.GetById(hbData.ClientId).State == State.offline ||
159          clientAdapter.GetById(hbData.ClientId).State == State.nullState) {
160        response.Success = false;
161        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_USER_NOT_LOGGED_IN;
162        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
163        return response;
164      }
165
166      if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
167        lastHeartbeats[hbData.ClientId] = DateTime.Now;
168      } else {
169        lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
170      }
171
172      response.Success = true;
173      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_HARDBEAT_RECEIVED;
174      List<Job> allOfflineJobs = new List<Job>(jobAdapter.GetJobsByState(State.offline));
175      if (allOfflineJobs.Count > 0 && hbData.freeCores > 0)
176        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.FetchJob));
177      else
178        response.ActionRequest.Add(new MessageContainer(MessageContainer.MessageType.NoMessage));
179
180      if (hbData.jobProgress != null) {
181        foreach (KeyValuePair<long, double> jobProgress in hbData.jobProgress) {
182          Job curJob = jobAdapter.GetById(jobProgress.Key);
183          curJob.Percentage = jobProgress.Value;
184          jobAdapter.Update(curJob);
185        }
186      }
187
188      return response;
189    }
190   
191    /// <summary>
192    /// if the client asked to pull a job he calls this method
193    /// the server selects a job and sends it to the client
194    /// </summary>
195    /// <param name="clientId"></param>
196    /// <returns></returns>
197    [MethodImpl(MethodImplOptions.Synchronized)]
198    public ResponseJob PullJob(Guid clientId) {
199      ResponseJob response = new ResponseJob();
200      lock (this) {
201        LinkedList<Job> allOfflineJobs = new LinkedList<Job>(jobAdapter.GetJobsByState(State.offline));
202        if (allOfflineJobs != null && allOfflineJobs.Count > 0) {
203          Job job2Calculate = allOfflineJobs.First.Value;
204          job2Calculate.State = State.calculating;
205          response.Job = job2Calculate;
206          jobAdapter.Update(job2Calculate);
207          response.Success = true;
208          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
209          return response;
210        }
211      }
212      response.Success = true;
213      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
214      return response;
215    }
216
217    /// <summary>
218    /// the client can send job results during calculating
219    /// and will send a final job result when he finished calculating
220    /// these job results will be stored in the database
221    /// </summary>
222    /// <param name="clientId"></param>
223    /// <param name="jobId"></param>
224    /// <param name="result"></param>
225    /// <param name="exception"></param>
226    /// <param name="finished"></param>
227    /// <returns></returns>
228    [MethodImpl(MethodImplOptions.Synchronized)]
229    public ResponseResultReceived SendJobResult(Guid clientId,
230      long jobId,
231      byte[] result,
232      Exception exception, 
233      bool finished) {
234      ResponseResultReceived response = new ResponseResultReceived();
235      ClientInfo client =
236        clientAdapter.GetById(clientId);
237
238      Job job =
239        jobAdapter.GetById(jobId);
240      if (job == null) {
241        response.Success = false;
242        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JO_WITH_THIS_ID;
243        return response;
244      }
245      if (job.State != State.calculating) {
246        response.Success = false;
247        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_WRONG_JOB_STATE;
248        return response;
249      }
250      if (finished) {
251        job.State = State.finished;
252        jobAdapter.Update(job);
253
254        List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
255        foreach (JobResult currentResult in jobResults)
256          jobResultAdapter.Delete(currentResult);
257      }
258
259      JobResult jobResult =
260        new JobResult();
261      jobResult.Client = client;
262      jobResult.Job = job;
263      jobResult.Result = result;
264      jobResult.Exception = exception;
265
266      jobResultAdapter.Update(jobResult);   
267
268      response.Success = true;
269      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOBRESULT_RECEIVED;
270      response.JobId = jobId;
271
272      return response;
273    }
274
275    /// <summary>
276    /// when a client logs out the state will be set
277    /// and the entry in the last hearbeats dictionary will be removed
278    /// </summary>
279    /// <param name="clientId"></param>
280    /// <returns></returns>
281    [MethodImpl(MethodImplOptions.Synchronized)]                       
282    public Response Logout(Guid clientId) {
283      Response response = new Response();
284
285      if (lastHeartbeats.ContainsKey(clientId))
286        lastHeartbeats.Remove(clientId);
287
288      ClientInfo client = clientAdapter.GetById(clientId);
289      if (client == null) {
290        response.Success = false;
291        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_CLIENT_NOT_REGISTERED;
292        return response;
293      }
294      client.State = State.offline;
295      clientAdapter.Update(client);
296
297      response.Success = true;
298      response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_LOGOUT_SUCCESS;
299     
300      return response;
301    }
302
303    #endregion
304  }
305}
Note: See TracBrowser for help on using the repository browser.