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

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

second fix (updated internal interfaces) (#529)

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