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

Last change on this file since 1715 was 1715, checked in by kgrading, 12 years ago

changed loading behavior back to byte array loading and removed dependency (#547)

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