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

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

part three of the awesome "FixTheBug!" Show (#529)

File size: 12.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.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
47
48namespace HeuristicLab.Hive.Client.Core {
49  /// <summary>
50  /// The core component of the Hive Client
51  /// </summary>
52  public class Core: MarshalByRefObject {
53    //Todo: Ask wagner
54    public static Object Locker { get; set; }   
55    public static bool abortRequested { get; set; }
56       
57    private Dictionary<long, Executor> engines = new Dictionary<long, Executor>();
58    private Dictionary<long, AppDomain> appDomains = new Dictionary<long, AppDomain>();
59    private Dictionary<long, Job> jobs = new Dictionary<long, Job>();
60
61    private WcfService wcfService;
62    private Heartbeat beat;
63   
64    /// <summary>
65    /// Main Method for the client
66    /// </summary>
67    public void Start() {
68      Core.Locker = new Object();
69      abortRequested = false;
70
71      Logging.GetInstance().Info(this.ToString(), "Hive Client started");
72      ClientConsoleServer server = new ClientConsoleServer();
73      server.StartClientConsoleServer(new Uri("net.tcp://127.0.0.1:8000/ClientConsole/"));
74
75      ConfigManager manager = ConfigManager.Instance;
76      manager.Core = this;
77     
78      //Register all Wcf Service references
79      wcfService = WcfService.Instance;
80      wcfService.LoginCompleted += new EventHandler<LoginCompletedEventArgs>(wcfService_LoginCompleted);
81      wcfService.SendJobCompleted += new EventHandler<SendJobCompletedEventArgs>(wcfService_PullJobCompleted);
82      wcfService.ProcessJobResultCompleted += new EventHandler<ProcessJobResultCompletedEventArgs>(wcfService_SendJobResultCompleted);
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.GetInstance().Info(this.ToString(), container.Message.ToString());
105        DetermineAction(container);
106      }
107    }
108
109    /// <summary>
110    /// Reads and analyzes the Messages from the MessageQueue and starts corresponding actions
111    /// </summary>
112    /// <param name="container">The Container, containing the message</param>
113    private void DetermineAction(MessageContainer container) {           
114      switch (container.Message) {
115        //Server requests to abort a job
116        case MessageContainer.MessageType.AbortJob:
117          engines[container.JobId].Abort();
118          break;
119        //Job has been successfully aborted
120        case MessageContainer.MessageType.JobAborted:
121          Debug.WriteLine("-- Job Aborted Message received");
122          break;
123        //Request a Snapshot from the Execution Engine
124        case MessageContainer.MessageType.RequestSnapshot:
125          engines[container.JobId].RequestSnapshot();
126          break;
127        //Snapshot is ready and can be sent back to the Server
128        case MessageContainer.MessageType.SnapshotReady:
129          ThreadPool.QueueUserWorkItem(new WaitCallback(GetSnapshot), container.JobId);
130          //Thread ssr = new Thread(new ParameterizedThreadStart(GetSnapshot));
131          //ssr.Start(container.JobId);         
132          break;
133        //Pull a Job from the Server
134        case MessageContainer.MessageType.FetchJob:
135          wcfService.SendJobAsync(ConfigManager.Instance.GetClientInfo().ClientId);
136          break;         
137        //A Job has finished and can be sent back to the server
138        case MessageContainer.MessageType.FinishedJob:
139          ThreadPool.QueueUserWorkItem(new WaitCallback(GetFinishedJob), container.JobId);
140          //Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
141          //finThread.Start(container.JobId);         
142          break;     
143        //Hard shutdown of the client
144        case MessageContainer.MessageType.Shutdown:
145          abortRequested = true;
146          beat.StopHeartBeat();
147          break;
148      }
149    }
150
151    //Asynchronous Threads for interaction with the Execution Engine
152    #region Async Threads for the EE
153   
154    private void GetFinishedJob(object jobId) {
155      long jId = (long)jobId;     
156      try {
157        byte[] sJob = engines[jId].GetFinishedJob();
158
159        if (WcfService.Instance.ConnState == NetworkEnum.WcfConnState.Loggedin) {
160          wcfService.ProcessJobResultAsync(ConfigManager.Instance.GetClientInfo().ClientId,
161            jId,
162            sJob,
163            1,
164            null,
165            true);
166        } else {         
167          JobStorageManager.PersistObjectToDisc(wcfService.ServerIP, wcfService.ServerPort, jId, sJob);
168          lock (Core.Locker) {
169            AppDomain.Unload(appDomains[jId]);
170            appDomains.Remove(jId);
171            engines.Remove(jId);
172            jobs.Remove(jId);
173          }
174        }
175      }
176      catch (InvalidStateException ise) {
177        Logging.GetInstance().Error(this.ToString(), "Exception: ", ise);
178      }
179    }
180
181    private void GetSnapshot(object jobId) {
182      long jId = (long)jobId;
183      byte[] obj = engines[jId].GetSnapshot();
184      wcfService.ProcessJobResultAsync(ConfigManager.Instance.GetClientInfo().ClientId,
185        jId,
186        obj,
187        engines[jId].Progress,
188        null,
189        false);
190    }
191
192    #endregion
193
194    //Eventhandlers for the communication with the wcf Layer
195    #region wcfService Events
196
197    void wcfService_LoginCompleted(object sender, LoginCompletedEventArgs e) {
198      if (e.Result.Success) {
199        Logging.GetInstance().Info(this.ToString(), "Login completed to Hive Server @ " + DateTime.Now);       
200      } else
201        Logging.GetInstance().Error(this.ToString(), e.Result.StatusMessage);
202    }   
203
204    void wcfService_PullJobCompleted(object sender, SendJobCompletedEventArgs e) {
205      if (e.Result.StatusMessage != ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT) {
206        bool sandboxed = true;
207
208        PluginManager.Manager.Initialize();
209        AppDomain appDomain = PluginManager.Manager.CreateAndInitAppDomainWithSandbox(e.Result.Job.Id.ToString(), sandboxed, typeof(TestJob));
210        appDomain.UnhandledException += new UnhandledExceptionEventHandler(appDomain_UnhandledException);
211        lock (Locker) {                   
212          if (!jobs.ContainsKey(e.Result.Job.Id)) {
213            jobs.Add(e.Result.Job.Id, e.Result.Job);
214            appDomains.Add(e.Result.Job.Id, appDomain);
215
216            Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
217            engine.JobId = e.Result.Job.Id;
218            engine.Queue = MessageQueue.GetInstance();
219            engine.Start(e.Result.Job.SerializedJob);
220            engines.Add(e.Result.Job.Id, engine);
221
222            ClientStatusInfo.JobsFetched++;
223
224            Debug.WriteLine("Increment FetchedJobs to:" + ClientStatusInfo.JobsFetched);
225          }
226        }
227      }
228    }
229   
230    //Todo: Seperate this method into 2: Finished jobs and Snapshots
231    void wcfService_SendJobResultCompleted(object sender, ProcessJobResultCompletedEventArgs e) {
232      if (e.Result.Success) {       
233        lock (Locker) {
234          //if the engine is running again -> we sent an snapshot. Otherwise the job was finished
235          //this method has a risk concerning race conditions.
236          //better expand the sendjobresultcompltedeventargs with a boolean "snapshot?" flag
237          if (e.Result.finished == false) {
238            Logging.GetInstance().Info(this.ToString(), "Snapshot for Job " + e.Result.JobId + " transmitted");
239          } else {
240            AppDomain.Unload(appDomains[e.Result.JobId]);
241            appDomains.Remove(e.Result.JobId);
242            engines.Remove(e.Result.JobId);
243            jobs.Remove(e.Result.JobId);
244            ClientStatusInfo.JobsProcessed++;
245            Debug.WriteLine("ProcessedJobs to:" + ClientStatusInfo.JobsProcessed);
246          }
247        }       
248      } else {
249        //Todo: don't Java Style! IT'S EVIL!
250        Logging.GetInstance().Error(this.ToString(), "Sending of job " + e.Result.JobId + " failed");
251      }
252    }
253
254    //Todo: First stop all threads, then terminate
255    void wcfService_ServerChanged(object sender, EventArgs e) {
256      Logging.GetInstance().Info(this.ToString(), "ServerChanged has been called");
257      lock (Locker) {
258        foreach (KeyValuePair<long, AppDomain> entries in appDomains)
259          AppDomain.Unload(appDomains[entries.Key]);
260        appDomains = new Dictionary<long, AppDomain>();
261        engines = new Dictionary<long, Executor>();
262      }
263    }
264
265    void wcfService_Connected(object sender, EventArgs e) {
266      wcfService.LoginSync(ConfigManager.Instance.GetClientInfo());
267      JobStorageManager.CheckAndSubmitJobsFromDisc();
268    }
269
270    //this is a little bit tricky -
271    void wcfService_ConnectionRestored(object sender, EventArgs e) {
272      Logging.GetInstance().Info(this.ToString(), "Reconnected to old server - checking currently running appdomains");                 
273
274      foreach (KeyValuePair<long, Executor> execKVP in engines) {
275        if (!execKVP.Value.Running && execKVP.Value.CurrentMessage == MessageContainer.MessageType.NoMessage) {
276          Logging.GetInstance().Info(this.ToString(), "Checking for JobId: " + execKVP.Value.JobId);
277          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
278          finThread.Start(execKVP.Value.JobId);
279        }
280      }
281    }
282
283    #endregion
284
285    public Dictionary<long, Executor> GetExecutionEngines() {
286      return engines;
287    }
288
289    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
290      Logging.GetInstance().Error(this.ToString(), " Exception: " + e.ExceptionObject.ToString());
291    }
292  }
293}
Note: See TracBrowser for help on using the repository browser.