Free cookie consent management tool by TermsFeed Policy Generator

source: trunk/sources/HeuristicLab.Hive.Client.Core/3.2/Core.cs @ 2063

Last change on this file since 2063 was 2063, checked in by kgrading, 15 years ago

added "time for shutdown" behavior: Snapshots of jobs get sent back to the server and then killed. When all jobs are killed, the client switches to disconnect state (#669)

File size: 16.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.Client.ExecutionEngine;
27using HeuristicLab.Hive.Client.Common;
28using System.Threading;
29using System.Reflection;
30using System.Diagnostics;
31using System.Security.Permissions;
32using System.Security.Policy;
33using System.Security;
34using HeuristicLab.Hive.Client.Communication;
35using HeuristicLab.Hive.Contracts.BusinessObjects;
36using HeuristicLab.Hive.Contracts;
37using System.Runtime.Remoting.Messaging;
38using HeuristicLab.PluginInfrastructure;
39using System.ServiceModel;
40using System.ServiceModel.Description;
41using HeuristicLab.Hive.Client.Core.ClientConsoleService;
42using HeuristicLab.Hive.Client.Core.ConfigurationManager;
43using HeuristicLab.Hive.Client.Communication.ServerService;
44using HeuristicLab.Hive.JobBase;
45using HeuristicLab.Hive.Client.Core.JobStorage;
46
47namespace HeuristicLab.Hive.Client.Core {
48  /// <summary>
49  /// The core component of the Hive Client
50  /// </summary>
51  public class Core: MarshalByRefObject {       
52    public static bool abortRequested { get; set; }
53    private bool currentlyFetching = false;
54
55    private Dictionary<Guid, Executor> engines = new Dictionary<Guid, Executor>();
56    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();
57    private Dictionary<Guid, Job> jobs = new Dictionary<Guid, Job>();
58
59    private WcfService wcfService;
60    private Heartbeat beat;
61   
62    /// <summary>
63    /// Main Method for the client
64    /// </summary>
65    public void Start() {     
66      abortRequested = false;
67      PluginManager.Manager.Initialize();
68      Logging.Instance.Info(this.ToString(), "Hive Client started");
69      ClientConsoleServer server = new ClientConsoleServer();
70      server.StartClientConsoleServer(new Uri("net.tcp://127.0.0.1:8000/ClientConsole/"));
71
72      ConfigManager manager = ConfigManager.Instance;
73      manager.Core = this;
74
75
76     
77      //Register all Wcf Service references
78      wcfService = WcfService.Instance;
79      wcfService.LoginCompleted += new EventHandler<LoginCompletedEventArgs>(wcfService_LoginCompleted);
80      wcfService.SendJobCompleted += new EventHandler<SendJobCompletedEventArgs>(wcfService_SendJobCompleted);
81      wcfService.StoreFinishedJobResultCompleted += new EventHandler<StoreFinishedJobResultCompletedEventArgs>(wcfService_StoreFinishedJobResultCompleted);
82      wcfService.ProcessSnapshotCompleted += new EventHandler<ProcessSnapshotCompletedEventArgs>(wcfService_ProcessSnapshotCompleted);
83      wcfService.ConnectionRestored += new EventHandler(wcfService_ConnectionRestored);
84      wcfService.ServerChanged += new EventHandler(wcfService_ServerChanged);
85      wcfService.Connected += new EventHandler(wcfService_Connected);
86      //Recover Server IP and Port from the Settings Framework
87      ConnectionContainer cc = ConfigManager.Instance.GetServerIPAndPort();     
88      if (cc.IPAdress != String.Empty && cc.Port != 0)
89        wcfService.SetIPAndPort(cc.IPAdress, cc.Port);
90
91      if (UptimeManager.Instance.isOnline())
92        wcfService.Connect();
93         
94      //Initialize the heartbeat
95      beat = new Heartbeat { Interval = 10000 };
96      beat.StartHeartbeat();     
97
98      MessageQueue queue = MessageQueue.GetInstance();
99     
100      //Main processing loop     
101      //Todo: own thread for message handling
102      //Rly?!
103      while (!abortRequested) {
104        MessageContainer container = queue.GetMessage();
105        Debug.WriteLine("Main loop received this message: " + container.Message.ToString());
106        Logging.Instance.Info(this.ToString(), container.Message.ToString());
107        DetermineAction(container);
108      }
109      Console.WriteLine("ended!");
110    }   
111
112    /// <summary>
113    /// Reads and analyzes the Messages from the MessageQueue and starts corresponding actions
114    /// </summary>
115    /// <param name="container">The Container, containing the message</param>
116    private void DetermineAction(MessageContainer container) {           
117      switch (container.Message) {
118        //Server requests to abort a job
119        case MessageContainer.MessageType.AbortJob:
120          if(engines.ContainsKey(container.JobId))
121            engines[container.JobId].Abort();
122          else
123            Logging.Instance.Error(this.ToString(), "AbortJob: Engine doesn't exist");
124          break;
125        //Job has been successfully aborted
126
127
128        case MessageContainer.MessageType.JobAborted:         
129        //todo: thread this
130          Debug.WriteLine("Job aborted, he's dead");
131          lock (engines) {           
132            Guid jobId = new Guid(container.JobId.ToString());
133            if(engines.ContainsKey(jobId)) {
134              appDomains[jobId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
135              AppDomain.Unload(appDomains[jobId]);
136              appDomains.Remove(jobId);
137              engines.Remove(jobId);
138              jobs.Remove(jobId);
139              GC.Collect();
140            } else
141              Logging.Instance.Error(this.ToString(), "JobAbort: Engine doesn't exist");
142          }
143          break;
144       
145       
146        //Request a Snapshot from the Execution Engine
147        case MessageContainer.MessageType.RequestSnapshot:
148          if (engines.ContainsKey(container.JobId))
149            engines[container.JobId].RequestSnapshot();
150          else
151            Logging.Instance.Error(this.ToString(), "RequestSnapshot: Engine doesn't exist");
152          break;
153       
154       
155        //Snapshot is ready and can be sent back to the Server
156        case MessageContainer.MessageType.SnapshotReady:
157          ThreadPool.QueueUserWorkItem(new WaitCallback(GetSnapshot), container.JobId);         
158          break;
159       
160       
161        //Pull a Job from the Server
162        case MessageContainer.MessageType.FetchJob:
163          if (!currentlyFetching) {
164            wcfService.SendJobAsync(ConfigManager.Instance.GetClientInfo().Id);
165            currentlyFetching = true;
166          }         
167          break;         
168       
169       
170        //A Job has finished and can be sent back to the server
171        case MessageContainer.MessageType.FinishedJob:
172          ThreadPool.QueueUserWorkItem(new WaitCallback(GetFinishedJob), container.JobId);         
173          break;     
174       
175       
176        case MessageContainer.MessageType.UptimeLimitDisconnect:
177          Logging.Instance.Info(this.ToString(), "Uptime Limit reached, storing jobs and sending them back");
178
179          //check if there are running jobs
180          if (engines.Count > 0) {
181            //make sure there is no more fetching of jobs while the snapshots get processed
182            currentlyFetching = true;
183            //request a snapshot of each running job
184            foreach (KeyValuePair<Guid, Executor> kvp in engines) {
185              kvp.Value.RequestSnapshot();
186            }
187           
188          } else {
189            //Disconnect afterwards
190            WcfService.Instance.Disconnect();
191          }
192          break;
193       
194       
195        //Hard shutdown of the client
196        case MessageContainer.MessageType.Shutdown:
197          lock (engines) {
198            foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
199              appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
200              AppDomain.Unload(kvp.Value);
201            }
202          }
203          abortRequested = true;
204          beat.StopHeartBeat();
205          WcfService.Instance.Logout(ConfigManager.Instance.GetClientInfo().Id);
206          break;
207      }
208    }
209
210    //Asynchronous Threads for interaction with the Execution Engine
211    #region Async Threads for the EE
212   
213    private void GetFinishedJob(object jobId) {
214      Guid jId = (Guid)jobId;     
215      try {
216        if (!engines.ContainsKey(jId)) {
217          Logging.Instance.Error(this.ToString(), "GetFinishedJob: Engine doesn't exist");
218          return;
219        }
220       
221        byte[] sJob = engines[jId].GetFinishedJob();
222
223        if (WcfService.Instance.ConnState == NetworkEnum.WcfConnState.Loggedin) {
224          wcfService.StoreFinishedJobResultAsync(ConfigManager.Instance.GetClientInfo().Id,
225            jId,
226            sJob,
227            1,
228            null,
229            true);
230        } else {
231          JobStorageManager.PersistObjectToDisc(wcfService.ServerIP, wcfService.ServerPort, jId, sJob);
232          lock (engines) {
233            appDomains[jId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
234            AppDomain.Unload(appDomains[jId]);
235            appDomains.Remove(jId);
236            engines.Remove(jId);
237            jobs.Remove(jId);
238          }
239        }
240      }
241      catch (InvalidStateException ise) {
242        Logging.Instance.Error(this.ToString(), "Exception: ", ise);
243      }
244    }
245
246    private void GetSnapshot(object jobId) {
247      Guid jId = (Guid)jobId;
248      byte[] obj = engines[jId].GetSnapshot();
249      wcfService.ProcessSnapshotSync(ConfigManager.Instance.GetClientInfo().Id,
250        jId,
251        obj,
252        engines[jId].Progress,
253        null);
254
255      //Uptime Limit reached, now is a good time to destroy this jobs.
256      if (!UptimeManager.Instance.isOnline()) {
257        lock (engines) {
258          appDomains[jId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
259          AppDomain.Unload(appDomains[jId]);
260          appDomains.Remove(jId);
261          engines.Remove(jId);
262          jobs.Remove(jId);
263        }
264        GC.Collect();
265
266        //Still anything running?
267        if (engines.Count == 0)
268          WcfService.Instance.Disconnect();
269     
270      } else {
271        engines[jId].StartOnlyJob();
272      }
273    }
274
275    #endregion
276
277    //Eventhandlers for the communication with the wcf Layer
278    #region wcfService Events
279
280    void wcfService_LoginCompleted(object sender, LoginCompletedEventArgs e) {
281      if (e.Result.Success) {
282        currentlyFetching = false;
283        Logging.Instance.Info(this.ToString(), "Login completed to Hive Server @ " + DateTime.Now);       
284      } else
285        Logging.Instance.Error(this.ToString(), e.Result.StatusMessage);
286    }   
287
288    void wcfService_SendJobCompleted(object sender, SendJobCompletedEventArgs e) {
289      if (e.Result.StatusMessage != ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT) {       
290        bool sandboxed = false;
291        //todo: For testing!!!
292        //beat.StopHeartBeat();       
293        //Todo: make a set & override the equals method
294        List<byte[]> files = new List<byte[]>();
295        foreach (CachedHivePluginInfo plugininfo in PluginCache.Instance.GetPlugins(e.Result.Job.PluginsNeeded))
296          files.AddRange(plugininfo.PluginFiles);
297       
298        AppDomain appDomain = PluginManager.Manager.CreateAndInitAppDomainWithSandbox(e.Result.Job.Id.ToString(), sandboxed, null, files);
299        appDomain.UnhandledException += new UnhandledExceptionEventHandler(appDomain_UnhandledException);
300        lock (engines) {                   
301          if (!jobs.ContainsKey(e.Result.Job.Id)) {
302            jobs.Add(e.Result.Job.Id, e.Result.Job);
303            appDomains.Add(e.Result.Job.Id, appDomain);
304
305            Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
306            engine.JobId = e.Result.Job.Id;
307            engine.Queue = MessageQueue.GetInstance();           
308            engine.Start(e.Result.Job.SerializedJob);
309            engines.Add(e.Result.Job.Id, engine);
310
311            ClientStatusInfo.JobsFetched++;
312
313            Debug.WriteLine("Increment FetchedJobs to:" + ClientStatusInfo.JobsFetched);
314          }
315        }       
316      }
317      currentlyFetching = false;
318    }
319
320    void wcfService_StoreFinishedJobResultCompleted(object sender, StoreFinishedJobResultCompletedEventArgs e) {
321      lock(engines) {
322        try {
323          appDomains[e.Result.JobId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
324          AppDomain.Unload(appDomains[e.Result.JobId]);
325          appDomains.Remove(e.Result.JobId);
326          engines.Remove(e.Result.JobId);
327          jobs.Remove(e.Result.JobId);
328        }
329        catch (Exception ex) {
330          Logging.Instance.Error(this.ToString(), "Exception when unloading the appdomain: ", ex);
331        }
332      }
333      if (e.Result.Success) {       
334     
335        //if the engine is running again -> we sent an snapshot. Otherwise the job was finished
336        //this method has a risk concerning race conditions.
337        //better expand the sendjobresultcompltedeventargs with a boolean "snapshot?" flag
338
339        ClientStatusInfo.JobsProcessed++;
340        Debug.WriteLine("ProcessedJobs to:" + ClientStatusInfo.JobsProcessed);               
341      } else {       
342        Logging.Instance.Error(this.ToString(), "Sending of job " + e.Result.JobId + " failed, job has been wasted. Message: " + e.Result.StatusMessage);
343      }
344    }
345
346    void wcfService_ProcessSnapshotCompleted(object sender, ProcessSnapshotCompletedEventArgs e) {
347      Logging.Instance.Info(this.ToString(), "Snapshot " + e.Result.JobId + " has been transmitted according to plan.");
348    }
349
350    //Todo: First stop all threads, then terminate
351    void wcfService_ServerChanged(object sender, EventArgs e) {
352      Logging.Instance.Info(this.ToString(), "ServerChanged has been called");
353      lock (engines) {
354        foreach (KeyValuePair<Guid, AppDomain> entries in appDomains) {
355          appDomains[entries.Key].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
356          AppDomain.Unload(appDomains[entries.Key]);
357        }
358        appDomains = new Dictionary<Guid, AppDomain>();
359        engines = new Dictionary<Guid, Executor>();
360      }
361    }
362
363    void wcfService_Connected(object sender, EventArgs e) {
364      wcfService.LoginSync(ConfigManager.Instance.GetClientInfo());
365      JobStorageManager.CheckAndSubmitJobsFromDisc();
366    }
367
368    //this is a little bit tricky -
369    void wcfService_ConnectionRestored(object sender, EventArgs e) {
370      Logging.Instance.Info(this.ToString(), "Reconnected to old server - checking currently running appdomains");                 
371
372      foreach (KeyValuePair<Guid, Executor> execKVP in engines) {
373        if (!execKVP.Value.Running && execKVP.Value.CurrentMessage == MessageContainer.MessageType.NoMessage) {
374          Logging.Instance.Info(this.ToString(), "Checking for JobId: " + execKVP.Value.JobId);
375          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
376          finThread.Start(execKVP.Value.JobId);
377        }
378      }
379    }
380
381    #endregion
382
383    public Dictionary<Guid, Executor> GetExecutionEngines() {
384      return engines;
385    }
386
387    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
388      Logging.Instance.Error(this.ToString(), "Exception in AppDomain: " + e.ExceptionObject.ToString());     
389    }
390
391    internal Dictionary<Guid, Job> GetJobs() {           
392      return jobs;
393    }
394  }
395}
Note: See TracBrowser for help on using the repository browser.