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

Last change on this file since 924 was 924, checked in by kgrading, 13 years ago

added a connectionRestored Event (#418)

File size: 6.8 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;
42
43
44namespace HeuristicLab.Hive.Client.Core {
45  public class Core {
46
47    public delegate string GetASnapshotDelegate();
48
49    Dictionary<long, Executor> engines = new Dictionary<long, Executor>();
50    Dictionary<long, AppDomain> appDomains = new Dictionary<long, AppDomain>();
51
52    private WcfService wcfService;
53
54    public void Start() {
55
56      ClientConsoleServer server = new ClientConsoleServer();
57      server.StartClientConsoleServer(new Uri("net.tcp://127.0.0.1:8000/ClientConsole/"));
58
59      ConfigurationManager manager = ConfigurationManager.GetInstance();
60      manager.Core = this;
61
62      wcfService = WcfService.Instance;
63      wcfService.Connect("192.168.132.1", "9000");
64
65      wcfService.LoginCompleted += new EventHandler<LoginCompletedEventArgs>(wcfService_LoginCompleted);
66      wcfService.PullJobCompleted += new EventHandler<PullJobCompletedEventArgs>(wcfService_PullJobCompleted);
67      wcfService.SendJobResultCompleted += new EventHandler<SendJobResultCompletedEventArgs>(wcfService_SendJobResultCompleted);
68      wcfService.ConnectionRestored += new EventHandler(wcfService_ConnectionRestored);
69
70      wcfService.LoginAsync(ConfigurationManager.GetInstance().GetClientInfo());
71
72      Heartbeat beat = new Heartbeat { Interval = 10000 };
73      beat.StartHeartbeat();     
74
75      MessageQueue queue = MessageQueue.GetInstance();
76      while (true) {
77        MessageContainer container = queue.GetMessage();
78        Debug.WriteLine("Main loop received this message: " + container.Message.ToString());
79        Logging.GetInstance().Info(this.ToString(), container.Message.ToString());
80        DetermineAction(container);
81      }
82    }
83
84    private void DetermineAction(MessageContainer container) {
85      switch (container.Message) {
86        case MessageContainer.MessageType.AbortJob:
87          engines[container.JobId].Abort();
88          break;
89        case MessageContainer.MessageType.JobAborted:
90          Debug.WriteLine("-- Job Aborted Message received");
91          break;
92
93        case MessageContainer.MessageType.RequestSnapshot:
94          engines[container.JobId].RequestSnapshot();
95          break;
96        case MessageContainer.MessageType.SnapshotReady:
97          Thread ssr = new Thread(new ParameterizedThreadStart(GetSnapshot));
98          ssr.Start(container.JobId);         
99          break;
100
101        case MessageContainer.MessageType.FetchJob:
102          wcfService.PullJobAsync(Guid.NewGuid());
103          break;         
104        case MessageContainer.MessageType.FinishedJob:
105          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
106          finThread.Start(container.JobId);         
107          break;     
108      }
109    }
110
111    #region Async Threads for the EE
112   
113    private void GetFinishedJob(object jobId) {
114      long jId = (long)jobId;
115      byte[] sJob = engines[jId].GetFinishedJob();
116     
117      JobResult jobResult = new JobResult { JobId = jId, Result = sJob, Client = ConfigurationManager.GetInstance().GetClientInfo() };
118      wcfService.SendJobResultAsync(jobResult, true);
119    }
120
121    private void GetSnapshot(object jobId) {
122      long jId = (long)jobId;
123      byte[] obj = engines[jId].GetSnapshot();
124    }
125
126    #endregion
127
128    #region wcfService Events
129
130    void wcfService_ConnectionRestored(object sender, EventArgs e) {
131      //Do some fancy new things here... e.g: check all appdomains if there are still active Jobs that need to be transmitted
132    }
133
134    void wcfService_LoginCompleted(object sender, LoginCompletedEventArgs e) {
135      if (e.Result.Success) {
136        Logging.GetInstance().Info(this.ToString(), "Login completed to Hive Server @ " + DateTime.Now);
137        ConfigurationManager.GetInstance().Loggedin();
138        Status.LoginTime = DateTime.Now;
139        Status.LoggedIn = true;
140      } else
141        Logging.GetInstance().Error(this.ToString(), e.Result.StatusMessage);
142    }   
143
144    void wcfService_PullJobCompleted(object sender, PullJobCompletedEventArgs e) {
145      bool sandboxed = false;
146
147      PluginManager.Manager.Initialize();
148      AppDomain appDomain =  PluginManager.Manager.CreateAndInitAppDomainWithSandbox(e.Result.JobId.ToString(), sandboxed, typeof(TestJob));
149     
150      appDomains.Add(e.Result.JobId, appDomain);
151
152      Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
153      engine.JobId = e.Result.JobId;
154      engine.Queue = MessageQueue.GetInstance();
155      engine.Start(e.Result.SerializedJob);
156      engines.Add(e.Result.JobId, engine);
157
158      Status.CurrentJobs++;
159
160      Debug.WriteLine("Increment CurrentJobs to:"+Status.CurrentJobs.ToString());
161    }
162
163    void wcfService_SendJobResultCompleted(object sender, SendJobResultCompletedEventArgs e) {
164      if (e.Result.Success) {
165        AppDomain.Unload(appDomains[e.Result.JobId]);
166        appDomains.Remove(e.Result.JobId);
167        engines.Remove(e.Result.JobId);
168        Status.CurrentJobs--;
169        Debug.WriteLine("Decrement CurrentJobs to:" + Status.CurrentJobs.ToString());
170      } else {
171        Debug.WriteLine("Job sending FAILED!");
172      }
173    }
174
175    #endregion
176
177    public Dictionary<long, Executor> GetExecutionEngines() {
178      return engines;
179    }
180  }
181}
Note: See TracBrowser for help on using the repository browser.