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

Last change on this file since 1120 was 1120, checked in by svonolfe, 12 years ago

Added execution engine facade (#465)

File size: 9.2 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;
45
46
47namespace HeuristicLab.Hive.Client.Core {
48  public class Core: MarshalByRefObject {
49    public delegate string GetASnapshotDelegate();
50
51    public static Object Locker { get; set; }
52
53    public static bool ShutdownFlag { get; set; }
54
55    Dictionary<long, Executor> engines = new Dictionary<long, Executor>();
56    Dictionary<long, AppDomain> appDomains = new Dictionary<long, AppDomain>();
57    Dictionary<long, Job> jobs = new Dictionary<long, Job>();
58
59    private WcfService wcfService;
60    private Heartbeat beat;
61
62    public void Start() {
63      Core.Locker = new Object();
64      ShutdownFlag = false;
65
66      Logging.GetInstance().Info(this.ToString(), "Hive Client started");
67      ClientConsoleServer server = new ClientConsoleServer();
68      server.StartClientConsoleServer(new Uri("net.tcp://127.0.0.1:8000/ClientConsole/"));
69
70      ConfigManager manager = ConfigManager.Instance;
71      manager.Core = this;
72
73      wcfService = WcfService.Instance;
74      wcfService.LoginCompleted += new EventHandler<LoginCompletedEventArgs>(wcfService_LoginCompleted);
75      wcfService.PullJobCompleted += new EventHandler<PullJobCompletedEventArgs>(wcfService_PullJobCompleted);
76      wcfService.SendJobResultCompleted += new EventHandler<SendJobResultCompletedEventArgs>(wcfService_SendJobResultCompleted);
77      wcfService.ConnectionRestored += new EventHandler(wcfService_ConnectionRestored);
78      wcfService.ServerChanged += new EventHandler(wcfService_ServerChanged);
79      wcfService.Connected += new EventHandler(wcfService_Connected);
80      ConnectionContainer cc = ConfigManager.Instance.GetServerIPAndPort();
81      if (cc.IPAdress != String.Empty && cc.Port != 0) {
82        wcfService.Connect(cc.IPAdress, cc.Port);
83      }
84   
85      beat = new Heartbeat { Interval = 10000 };
86      beat.StartHeartbeat();     
87
88      MessageQueue queue = MessageQueue.GetInstance();
89      while (!ShutdownFlag) {
90        MessageContainer container = queue.GetMessage();
91        Debug.WriteLine("Main loop received this message: " + container.Message.ToString());
92        Logging.GetInstance().Info(this.ToString(), container.Message.ToString());
93        DetermineAction(container);
94      }
95    }
96
97    private void DetermineAction(MessageContainer container) {
98      switch (container.Message) {
99        case MessageContainer.MessageType.AbortJob:
100          engines[container.JobId].Abort();
101          break;
102        case MessageContainer.MessageType.JobAborted:
103          Debug.WriteLine("-- Job Aborted Message received");
104          break;
105
106        case MessageContainer.MessageType.RequestSnapshot:
107          engines[container.JobId].RequestSnapshot();
108          break;
109        case MessageContainer.MessageType.SnapshotReady:
110          Thread ssr = new Thread(new ParameterizedThreadStart(GetSnapshot));
111          ssr.Start(container.JobId);         
112          break;
113
114        case MessageContainer.MessageType.FetchJob:
115          wcfService.PullJobAsync(Guid.NewGuid());
116          break;         
117        case MessageContainer.MessageType.FinishedJob:
118          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
119          finThread.Start(container.JobId);         
120          break;     
121        case MessageContainer.MessageType.Shutdown:
122          ShutdownFlag = true;
123          beat.StopHeartBeat();
124          break;
125      }
126    }
127
128    #region Async Threads for the EE
129   
130    private void GetFinishedJob(object jobId) {
131      long jId = (long)jobId;
132      byte[] sJob = engines[jId].GetFinishedJob();
133
134      wcfService.SendJobResultAsync(ConfigManager.Instance.GetClientInfo().ClientId,
135        jId,
136        sJob,
137        null,
138        true);
139    }
140
141    private void GetSnapshot(object jobId) {
142      long jId = (long)jobId;
143      byte[] obj = engines[jId].GetSnapshot();
144    }
145
146    #endregion
147
148    #region wcfService Events
149
150    void wcfService_LoginCompleted(object sender, LoginCompletedEventArgs e) {
151      if (e.Result.Success) {
152        Logging.GetInstance().Info(this.ToString(), "Login completed to Hive Server @ " + DateTime.Now);       
153      } else
154        Logging.GetInstance().Error(this.ToString(), e.Result.StatusMessage);
155    }   
156
157    void wcfService_PullJobCompleted(object sender, PullJobCompletedEventArgs e) {
158      if (e.Result.StatusMessage != ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT) {
159        bool sandboxed = false;
160
161        PluginManager.Manager.Initialize();
162        AppDomain appDomain = PluginManager.Manager.CreateAndInitAppDomainWithSandbox(e.Result.Job.Id.ToString(), sandboxed, typeof(TestJob));
163        appDomain.UnhandledException += new UnhandledExceptionEventHandler(appDomain_UnhandledException);
164        lock (Locker) {                   
165          if (!jobs.ContainsKey(e.Result.Job.Id)) {
166            jobs.Add(e.Result.Job.Id, e.Result.Job);
167            appDomains.Add(e.Result.Job.Id, appDomain);
168
169            Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
170            engine.JobId = e.Result.Job.Id;
171            engine.Queue = MessageQueue.GetInstance();
172            engine.Start(e.Result.Job.SerializedJob);
173            engines.Add(e.Result.Job.Id, engine);
174
175            ClientStatusInfo.JobsFetched++;
176
177            Debug.WriteLine("Increment FetchedJobs to:" + ClientStatusInfo.JobsFetched);
178          }
179        }
180      }
181    }
182
183    void wcfService_SendJobResultCompleted(object sender, SendJobResultCompletedEventArgs e) {
184      if (e.Result.Success) {
185        lock (Locker) {
186          AppDomain.Unload(appDomains[e.Result.JobId]);
187          appDomains.Remove(e.Result.JobId);
188          engines.Remove(e.Result.JobId);
189          jobs.Remove(e.Result.JobId);
190          ClientStatusInfo.JobsProcessed++;
191        }
192        Debug.WriteLine("ProcessedJobs to:" + ClientStatusInfo.JobsProcessed);
193      } else {
194        Debug.WriteLine("Job sending FAILED!");
195      }
196    }
197
198    void wcfService_ServerChanged(object sender, EventArgs e) {
199      Logging.GetInstance().Info(this.ToString(), "ServerChanged has been called");
200      lock (Locker) {
201        foreach (KeyValuePair<long, AppDomain> entries in appDomains)
202          AppDomain.Unload(appDomains[entries.Key]);
203        appDomains = new Dictionary<long, AppDomain>();
204        engines = new Dictionary<long, Executor>();
205      }
206    }
207
208    void wcfService_Connected(object sender, EventArgs e) {
209      wcfService.LoginSync(ConfigManager.Instance.GetClientInfo());
210    }
211
212    //this is a little bit tricky -
213    void wcfService_ConnectionRestored(object sender, EventArgs e) {
214      Logging.GetInstance().Info(this.ToString(), "Reconnected to old server - checking currently running appdomains");                 
215
216      foreach (KeyValuePair<long, Executor> execKVP in engines) {
217        if (!execKVP.Value.Running && execKVP.Value.CurrentMessage == MessageContainer.MessageType.NoMessage) {
218          Logging.GetInstance().Info(this.ToString(), "Checking for JobId: " + execKVP.Value.JobId);
219          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
220          finThread.Start(execKVP.Value.JobId);
221        }
222      }
223    }
224
225    #endregion
226
227    public Dictionary<long, Executor> GetExecutionEngines() {
228      return engines;
229    }
230
231    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
232      Logging.GetInstance().Error(this.ToString(), " Exception: " + e.ExceptionObject.ToString());
233    }
234  }
235}
Note: See TracBrowser for help on using the repository browser.