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

Last change on this file since 1990 was 1990, checked in by gkronber, 12 years ago
  • Refactoring: renamed method LoadPlugins to LoadAssemblies in class Runner.
  • Added cache of loaded assemblies in the Runner.
  • Added an AssemblyResolveEvent that returns already loaded assemblies from the cache.

#658 (For the execution of jobs assemblies have to be loaded dynamically in the correct order)

File size: 14.5 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.Connect(cc.IPAdress, cc.Port);
90      }
91   
92      //Initialize the heartbeat
93      beat = new Heartbeat { Interval = 10000 };
94      beat.StartHeartbeat();     
95
96      MessageQueue queue = MessageQueue.GetInstance();
97     
98      //Main processing loop     
99      //Todo: own thread for message handling
100      //Rly?!
101      while (!abortRequested) {
102        MessageContainer container = queue.GetMessage();
103        Debug.WriteLine("Main loop received this message: " + container.Message.ToString());
104        Logging.Instance.Info(this.ToString(), container.Message.ToString());
105        DetermineAction(container);
106      }
107      Console.WriteLine("ended!");
108    }   
109
110    /// <summary>
111    /// Reads and analyzes the Messages from the MessageQueue and starts corresponding actions
112    /// </summary>
113    /// <param name="container">The Container, containing the message</param>
114    private void DetermineAction(MessageContainer container) {           
115      switch (container.Message) {
116        //Server requests to abort a job
117        case MessageContainer.MessageType.AbortJob:
118          if(engines.ContainsKey(container.JobId))
119            engines[container.JobId].Abort();
120          else
121            Logging.Instance.Error(this.ToString(), "AbortJob: Engine doesn't exist");
122          break;
123        //Job has been successfully aborted
124        case MessageContainer.MessageType.JobAborted:
125          //todo: thread this
126          Debug.WriteLine("Job aborted, he's dead");
127          lock (engines) {           
128            Guid jobId = new Guid(container.JobId.ToString());
129            if(engines.ContainsKey(jobId)) {
130              appDomains[jobId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
131              AppDomain.Unload(appDomains[jobId]);
132              appDomains.Remove(jobId);
133              engines.Remove(jobId);
134              jobs.Remove(jobId);
135              GC.Collect();
136            } else
137              Logging.Instance.Error(this.ToString(), "JobAbort: Engine doesn't exist");
138          }
139          break;
140        //Request a Snapshot from the Execution Engine
141        case MessageContainer.MessageType.RequestSnapshot:
142          if (engines.ContainsKey(container.JobId))
143            engines[container.JobId].RequestSnapshot();
144          else
145            Logging.Instance.Error(this.ToString(), "RequestSnapshot: Engine doesn't exist");
146          break;
147        //Snapshot is ready and can be sent back to the Server
148        case MessageContainer.MessageType.SnapshotReady:
149          ThreadPool.QueueUserWorkItem(new WaitCallback(GetSnapshot), container.JobId);         
150          break;
151        //Pull a Job from the Server
152        case MessageContainer.MessageType.FetchJob:
153          if (!currentlyFetching) {
154            wcfService.SendJobAsync(ConfigManager.Instance.GetClientInfo().Id);
155            currentlyFetching = true;
156          }         
157          break;         
158        //A Job has finished and can be sent back to the server
159        case MessageContainer.MessageType.FinishedJob:
160          ThreadPool.QueueUserWorkItem(new WaitCallback(GetFinishedJob), container.JobId);         
161          break;     
162        //Hard shutdown of the client
163        case MessageContainer.MessageType.Shutdown:
164          lock (engines) {
165            foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
166              appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
167              AppDomain.Unload(kvp.Value);
168            }
169          }
170          abortRequested = true;
171          beat.StopHeartBeat();
172          WcfService.Instance.Logout(ConfigManager.Instance.GetClientInfo().Id);
173          break;
174      }
175    }
176
177    //Asynchronous Threads for interaction with the Execution Engine
178    #region Async Threads for the EE
179   
180    private void GetFinishedJob(object jobId) {
181      Guid jId = (Guid)jobId;     
182      try {
183        if (!engines.ContainsKey(jId)) {
184          Logging.Instance.Error(this.ToString(), "GetFinishedJob: Engine doesn't exist");
185          return;
186        }
187       
188        byte[] sJob = engines[jId].GetFinishedJob();
189
190        if (WcfService.Instance.ConnState == NetworkEnum.WcfConnState.Loggedin) {
191          wcfService.StoreFinishedJobResultAsync(ConfigManager.Instance.GetClientInfo().Id,
192            jId,
193            sJob,
194            1,
195            null,
196            true);
197        } else {
198          JobStorageManager.PersistObjectToDisc(wcfService.ServerIP, wcfService.ServerPort, jId, sJob);
199          lock (engines) {
200            appDomains[jId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
201            AppDomain.Unload(appDomains[jId]);
202            appDomains.Remove(jId);
203            engines.Remove(jId);
204            jobs.Remove(jId);
205          }
206        }
207      }
208      catch (InvalidStateException ise) {
209        Logging.Instance.Error(this.ToString(), "Exception: ", ise);
210      }
211    }
212
213    private void GetSnapshot(object jobId) {
214      Guid jId = (Guid)jobId;
215      byte[] obj = engines[jId].GetSnapshot();
216      wcfService.ProcessSnapshotSync(ConfigManager.Instance.GetClientInfo().Id,
217        jId,
218        obj,
219        engines[jId].Progress,
220        null);
221      engines[jId].StartOnlyJob();
222    }
223
224    #endregion
225
226    //Eventhandlers for the communication with the wcf Layer
227    #region wcfService Events
228
229    void wcfService_LoginCompleted(object sender, LoginCompletedEventArgs e) {
230      if (e.Result.Success) {
231        Logging.Instance.Info(this.ToString(), "Login completed to Hive Server @ " + DateTime.Now);       
232      } else
233        Logging.Instance.Error(this.ToString(), e.Result.StatusMessage);
234    }   
235
236    void wcfService_SendJobCompleted(object sender, SendJobCompletedEventArgs e) {
237      if (e.Result.StatusMessage != ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT) {       
238        bool sandboxed = false;
239        //todo: For testing!!!
240        //beat.StopHeartBeat();       
241        //Todo: make a set & override the equals method
242        List<byte[]> files = new List<byte[]>();
243        foreach (CachedHivePluginInfo plugininfo in PluginCache.Instance.GetPlugins(e.Result.Job.PluginsNeeded))
244          files.AddRange(plugininfo.PluginFiles);
245       
246        AppDomain appDomain = PluginManager.Manager.CreateAndInitAppDomainWithSandbox(e.Result.Job.Id.ToString(), sandboxed, null, files);
247        appDomain.UnhandledException += new UnhandledExceptionEventHandler(appDomain_UnhandledException);
248        lock (engines) {                   
249          if (!jobs.ContainsKey(e.Result.Job.Id)) {
250            jobs.Add(e.Result.Job.Id, e.Result.Job);
251            appDomains.Add(e.Result.Job.Id, appDomain);
252
253            Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
254            engine.JobId = e.Result.Job.Id;
255            engine.Queue = MessageQueue.GetInstance();           
256            engine.Start(e.Result.Job.SerializedJob);
257            engines.Add(e.Result.Job.Id, engine);
258
259            ClientStatusInfo.JobsFetched++;
260
261            Debug.WriteLine("Increment FetchedJobs to:" + ClientStatusInfo.JobsFetched);
262          }
263        }       
264      }
265      currentlyFetching = false;
266    }
267
268    void wcfService_StoreFinishedJobResultCompleted(object sender, StoreFinishedJobResultCompletedEventArgs e) {
269      lock(engines) {
270        try {
271          appDomains[e.Result.JobId].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
272          AppDomain.Unload(appDomains[e.Result.JobId]);
273          appDomains.Remove(e.Result.JobId);
274          engines.Remove(e.Result.JobId);
275          jobs.Remove(e.Result.JobId);
276        }
277        catch (Exception ex) {
278          Logging.Instance.Error(this.ToString(), "Exception when unloading the appdomain: ", ex);
279        }
280      }
281      if (e.Result.Success) {       
282     
283        //if the engine is running again -> we sent an snapshot. Otherwise the job was finished
284        //this method has a risk concerning race conditions.
285        //better expand the sendjobresultcompltedeventargs with a boolean "snapshot?" flag
286
287        ClientStatusInfo.JobsProcessed++;
288        Debug.WriteLine("ProcessedJobs to:" + ClientStatusInfo.JobsProcessed);               
289      } else {       
290        Logging.Instance.Error(this.ToString(), "Sending of job " + e.Result.JobId + " failed, job has been wasted. Message: " + e.Result.StatusMessage);
291      }
292    }
293
294    void wcfService_ProcessSnapshotCompleted(object sender, ProcessSnapshotCompletedEventArgs e) {
295      Logging.Instance.Info(this.ToString(), "Snapshot " + e.Result.JobId + " has been transmitted according to plan.");
296    }
297
298    //Todo: First stop all threads, then terminate
299    void wcfService_ServerChanged(object sender, EventArgs e) {
300      Logging.Instance.Info(this.ToString(), "ServerChanged has been called");
301      lock (engines) {
302        foreach (KeyValuePair<Guid, AppDomain> entries in appDomains) {
303          appDomains[entries.Key].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
304          AppDomain.Unload(appDomains[entries.Key]);
305        }
306        appDomains = new Dictionary<Guid, AppDomain>();
307        engines = new Dictionary<Guid, Executor>();
308      }
309    }
310
311    void wcfService_Connected(object sender, EventArgs e) {
312      wcfService.LoginSync(ConfigManager.Instance.GetClientInfo());
313      JobStorageManager.CheckAndSubmitJobsFromDisc();
314    }
315
316    //this is a little bit tricky -
317    void wcfService_ConnectionRestored(object sender, EventArgs e) {
318      Logging.Instance.Info(this.ToString(), "Reconnected to old server - checking currently running appdomains");                 
319
320      foreach (KeyValuePair<Guid, Executor> execKVP in engines) {
321        if (!execKVP.Value.Running && execKVP.Value.CurrentMessage == MessageContainer.MessageType.NoMessage) {
322          Logging.Instance.Info(this.ToString(), "Checking for JobId: " + execKVP.Value.JobId);
323          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
324          finThread.Start(execKVP.Value.JobId);
325        }
326      }
327    }
328
329    #endregion
330
331    public Dictionary<Guid, Executor> GetExecutionEngines() {
332      return engines;
333    }
334
335    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
336      Logging.Instance.Error(this.ToString(), "Exception in AppDomain: " + e.ExceptionObject.ToString());     
337    }
338
339    internal Dictionary<Guid, Job> GetJobs() {           
340      return jobs;
341    }
342  }
343}
Note: See TracBrowser for help on using the repository browser.