source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Core.cs @ 5778

Last change on this file since 5778 was 5778, checked in by ascheibe, 11 years ago

#1233

  • log uncaught exceptions to an eventlog if available
  • fixed job pause bug
File size: 20.0 KB
Line 
1#region License Information
2/* HeuristicLab
3 * Copyright (C) 2002-2010 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.Diagnostics;
25using System.IO;
26using System.Runtime.CompilerServices;
27using System.ServiceModel;
28using System.Threading;
29using System.Threading.Tasks;
30using HeuristicLab.Clients.Hive.SlaveCore.ServiceContracts;
31using HeuristicLab.Common;
32using HeuristicLab.Core;
33
34
35
36namespace HeuristicLab.Clients.Hive.SlaveCore {
37  /// <summary>
38  /// The core component of the Hive Client
39  /// </summary>
40  public class Core : MarshalByRefObject {
41
42    //TODO: this class should be a singleton; there is only one instance, the reference is meanwhile save in TheCore
43    public static Core TheCore;
44
45    public EventLog ServiceEventLog { get; set; }
46
47    public static bool abortRequested { get; set; }
48    private Semaphore waitShutdownSem = new Semaphore(0, 1);
49    public static ILog Log { get; set; }
50
51    private Dictionary<Guid, Executor> engines = new Dictionary<Guid, Executor>();
52    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();
53    private Dictionary<Guid, Job> jobs = new Dictionary<Guid, Job>();
54
55    private WcfService wcfService;
56    private HeartbeatManager heartbeatManager;
57    private int coreThreadId;
58
59    private ISlaveCommunication ClientCom;
60    private ServiceHost slaveComm;
61
62    public Dictionary<Guid, Executor> ExecutionEngines {
63      get { return engines; }
64    }
65
66    internal Dictionary<Guid, Job> Jobs {
67      get { return jobs; }
68    }
69
70    public Core() {
71      TheCore = this;
72    }
73
74    /// <summary>
75    /// Main Method for the client
76    /// </summary>
77    public void Start() {
78      coreThreadId = Thread.CurrentThread.ManagedThreadId;
79      abortRequested = false;
80
81      try {
82        //start the client communication service (pipe between slave and slave gui)
83        slaveComm = new ServiceHost(typeof(SlaveCommunicationService));
84        slaveComm.Open();
85
86        ClientCom = SlaveClientCom.Instance.ClientCom;
87        ClientCom.LogMessage("Hive Slave started");
88
89        ConfigManager manager = ConfigManager.Instance;
90        manager.Core = this;
91
92        wcfService = WcfService.Instance;
93        RegisterServiceEvents();
94
95        StartHeartbeats(); // Start heartbeats thread
96        DispatchMessageQueue(); // dispatch messages until abortRequested
97      }
98      catch (Exception ex) {
99        if (ServiceEventLog != null) {
100          try {
101            ServiceEventLog.WriteEntry("Hive Slave threw exception: " + ex.ToString() + " with stack trace: " + ex.StackTrace);
102          }
103          catch (Exception) { }
104        } else {
105          throw ex;
106        }
107      }
108      finally {
109        DeRegisterServiceEvents();
110        waitShutdownSem.Release();
111      }
112    }
113
114    private void StartHeartbeats() {
115      //Initialize the heartbeat     
116      if (heartbeatManager != null) {
117        heartbeatManager = new HeartbeatManager { Interval = new TimeSpan(0, 0, 10) };
118        heartbeatManager.StartHeartbeat();
119      }
120    }
121
122    private void DispatchMessageQueue() {
123      MessageQueue queue = MessageQueue.GetInstance();
124      while (!abortRequested) {
125        MessageContainer container = queue.GetMessage();
126        DetermineAction(container);
127      }
128    }
129
130    private void RegisterServiceEvents() {
131      WcfService.Instance.Connected += new EventHandler(WcfService_Connected);
132      WcfService.Instance.ExceptionOccured += new EventHandler<EventArgs<Exception>>(WcfService_ExceptionOccured);
133    }
134
135    private void DeRegisterServiceEvents() {
136      WcfService.Instance.Connected -= WcfService_Connected;
137      WcfService.Instance.ExceptionOccured -= WcfService_ExceptionOccured;
138    }
139
140    void WcfService_ExceptionOccured(object sender, EventArgs<Exception> e) {
141      ClientCom.LogMessage("Connection to server interruped with exception: " + e.Value.Message);
142    }
143
144    void WcfService_Connected(object sender, EventArgs e) {
145      ClientCom.LogMessage("Connected successfully to Hive server");
146    }
147
148    /// <summary>
149    /// Reads and analyzes the Messages from the MessageQueue and starts corresponding actions
150    /// </summary>
151    /// <param name="container">The Container, containing the message</param>
152    private void DetermineAction(MessageContainer container) {
153      ClientCom.LogMessage("Message: " + container.Message.ToString() + " for job: " + container.JobId);
154
155      if (container is ExecutorMessageContainer<Guid>) {
156        ExecutorMessageContainer<Guid> c = (ExecutorMessageContainer<Guid>)container;
157        c.execute();
158      } else if (container is MessageContainer) {
159        switch (container.Message) {
160          case MessageContainer.MessageType.CalculateJob:
161            Task.Factory.StartNew(() => {
162              Job job = wcfService.GetJob(container.JobId);
163              lock (engines) {
164                if (!jobs.ContainsKey(job.Id)) {
165                  jobs.Add(job.Id, job);
166                }
167              }
168              JobData jobData = wcfService.GetJobData(job.Id);
169              job = wcfService.UpdateJobState(job.Id, JobState.Calculating, null);
170              StartJobInAppDomain(job, jobData);
171            });
172            break;
173          case MessageContainer.MessageType.ShutdownSlave:
174            ShutdownCore();
175            break;
176          case MessageContainer.MessageType.StopAll:
177            DoStopAll();
178            break;
179          case MessageContainer.MessageType.PauseAll:
180            DoPauseAll();
181            break;
182          case MessageContainer.MessageType.AbortAll:
183            DoAbortAll();
184            break;
185          case MessageContainer.MessageType.AbortJob:
186            KillAppDomain(container.JobId);
187            break;
188          case MessageContainer.MessageType.StopJob:
189            DoStopJob(container.JobId);
190            break;
191          case MessageContainer.MessageType.PauseJob:
192            DoPauseJob(container.JobId);
193            break;
194          case MessageContainer.MessageType.Restart:
195            DoStartSlave();
196            break;
197          case MessageContainer.MessageType.Sleep:
198            Sleep();
199            break;
200          case MessageContainer.MessageType.SayHello:
201            wcfService.Connect(ConfigManager.Instance.GetClientInfo());
202            break;
203        }
204      } else {
205        ClientCom.LogMessage("Unknown MessageContainer: " + container);
206      }
207    }
208
209    private void DoPauseJob(Guid jobId) {
210      Job job = Jobs[jobId];
211
212      if (job != null) {
213        engines[job.Id].Pause();
214        JobData sJob = engines[job.Id].GetPausedJob();
215        // job.Exception = engines[job.Id].CurrentException; // can there be an exception if a job is paused
216        job.ExecutionTime = engines[job.Id].ExecutionTime;
217
218        try {
219          ClientCom.LogMessage("Sending the paused job with id: " + job.Id);
220          wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
221          SlaveStatusInfo.JobsProcessed++;    //TODO: count or not count, thats the question
222        }
223        catch (Exception e) {
224          ClientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
225        }
226        finally {
227          KillAppDomain(job.Id); // kill app-domain in every case         
228        }
229      }
230    }
231
232    private void DoStopJob(Guid guid) {
233      Job job = Jobs[guid];
234
235      if (job != null) {
236        engines[job.Id].Stop();
237        JobData sJob = engines[job.Id].GetFinishedJob();
238        // job.Exception = engines[job.Id].CurrentException; // can there be an exception if a job is stopped regularly
239        job.ExecutionTime = engines[job.Id].ExecutionTime;
240
241        try {
242          ClientCom.LogMessage("Sending the stoppped job with id: " + job.Id);
243          wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
244          SlaveStatusInfo.JobsProcessed++;    //TODO: count or not count, thats the question
245        }
246        catch (Exception e) {
247          ClientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
248        }
249        finally {
250          KillAppDomain(job.Id); // kill app-domain in every case         
251        }
252      }
253    }
254
255    /// <summary>
256    /// aborts all running jobs, no results are sent back
257    /// </summary>
258    private void DoAbortAll() {
259      List<Guid> guids = new List<Guid>();
260      foreach (Guid job in Jobs.Keys) {
261        guids.Add(job);
262      }
263
264      foreach (Guid g in guids) {
265        KillAppDomain(g);
266      }
267
268      ClientCom.LogMessage("Aborted all jobs!");
269    }
270
271    /// <summary>
272    /// wait for jobs to finish, then pause client
273    /// </summary>
274    private void DoPauseAll() {
275      ClientCom.LogMessage("Pause all received");
276
277      //copy guids because there will be removed items from 'Jobs'
278      List<Guid> guids = new List<Guid>();
279      foreach (Guid job in Jobs.Keys) {
280        guids.Add(job);
281      }
282
283      foreach (Guid g in guids) {
284        DoPauseJob(g);
285      }
286    }
287
288    /// <summary>
289    /// pause slave immediately
290    /// </summary>
291    private void DoStopAll() {
292      ClientCom.LogMessage("Stop all received");
293
294      //copy guids because there will be removed items from 'Jobs'
295      List<Guid> guids = new List<Guid>();
296      foreach (Guid job in Jobs.Keys) {
297        guids.Add(job);
298      }
299
300      foreach (Guid g in guids) {
301        DoStopJob(g);
302      }
303    }
304
305    /// <summary>
306    /// completly shudown slave
307    /// </summary>
308    public void Shutdown() {
309      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.ShutdownSlave);
310      MessageQueue.GetInstance().AddMessage(mc);
311      waitShutdownSem.WaitOne();
312    }
313
314    /// <summary>
315    /// complete shutdown, should be called before the the application is exited
316    /// </summary>
317    private void ShutdownCore() {
318      ClientCom.LogMessage("Shutdown Signal received");
319      ClientCom.LogMessage("Stopping heartbeat");
320      heartbeatManager.StopHeartBeat();
321      abortRequested = true;
322      ClientCom.LogMessage("Logging out");
323
324
325      lock (engines) {
326        ClientCom.LogMessage("engines locked");
327        foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
328          ClientCom.LogMessage("Shutting down Appdomain for " + kvp.Key);
329          appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
330          AppDomain.Unload(kvp.Value);
331        }
332      }
333      WcfService.Instance.Disconnect();
334      ClientCom.Shutdown();
335      SlaveClientCom.Close();
336
337      if (slaveComm.State != CommunicationState.Closed)
338        slaveComm.Close();
339    }
340
341    /// <summary>
342    /// reinitializes everything and continues operation,
343    /// can be called after Sleep()
344    /// </summary> 
345    private void DoStartSlave() {
346      ClientCom.LogMessage("Restart received");
347      StartHeartbeats();
348      ClientCom.LogMessage("Restart done");
349    }
350
351    /// <summary>
352    /// stop slave, except for client gui communication,
353    /// primarily used by gui if core is running as windows service
354    /// </summary>
355    //TODO: do we need an AbortSleep?
356    private void Sleep() {
357      ClientCom.LogMessage("Sleep received");
358      heartbeatManager.StopHeartBeat();
359      heartbeatManager = null;
360      DoStopAll();
361      WcfService.Instance.Disconnect();
362      ClientCom.LogMessage("Sleep done");
363    }
364
365    /// <summary>
366    /// Pauses a job, which means sending it to the server and killing it locally;
367    /// atm only used when executor is waiting for child jobs
368    /// </summary>
369    /// <param name="data"></param>
370    [MethodImpl(MethodImplOptions.Synchronized)]
371    public void PauseWaitJob(JobData data) {
372      if (!Jobs.ContainsKey(data.JobId)) {
373        ClientCom.LogMessage("Can't find job with id " + data.JobId);
374      } else {
375        Job job = Jobs[data.JobId];
376        wcfService.UpdateJobData(job, data, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
377        wcfService.UpdateJobState(job.Id, JobState.Waiting, null);
378      }
379      KillAppDomain(data.JobId);
380    }
381
382    /// <summary>
383    /// serializes the finished job and submits it to the server. If, at the time, a network connection is unavailable, the Job gets stored on the disk.
384    /// once the connection gets reestablished, the job gets submitted
385    /// </summary>
386    /// <param name="jobId"></param>
387    [MethodImpl(MethodImplOptions.Synchronized)]
388    public void SendFinishedJob(Guid jobId) {
389      try {
390        ClientCom.LogMessage("Getting the finished job with id: " + jobId);
391        if (!engines.ContainsKey(jobId)) {
392          ClientCom.LogMessage("Engine doesn't exist");
393          return;
394        }
395        if (!jobs.ContainsKey(jobId)) {
396          ClientCom.LogMessage("Job doesn't exist");
397          return;
398        }
399        Job cJob = jobs[jobId];
400        cJob.ExecutionTime = engines[jobId].ExecutionTime;
401
402        JobData sJob = engines[jobId].GetFinishedJob();
403        // cJob.Exception = engines[jId].CurrentException; // can there be an exception if the job is sent normally. the exception should be entered in the statelog with the corresponding state (Failed)
404        cJob.ExecutionTime = engines[jobId].ExecutionTime;
405
406        try {
407          ClientCom.LogMessage("Sending the finished job with id: " + jobId);
408          wcfService.UpdateJobData(cJob, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Finished);
409          SlaveStatusInfo.JobsProcessed++;
410        }
411        catch (Exception e) {
412          ClientCom.LogMessage("Transmitting to server failed. Storing the finished job with id: " + jobId + " to hdd (" + e.ToString() + ")");
413        }
414        finally {
415          KillAppDomain(jobId); // kill app-domain in every case
416          heartbeatManager.AwakeHeartBeatThread();
417        }
418      }
419      catch (Exception e) {
420        OnExceptionOccured(e);
421      }
422    }
423
424    /// <summary>
425    /// A new Job from the wcfService has been received and will be started within a AppDomain.
426    /// </summary>
427    /// <param name="sender"></param>
428    /// <param name="e"></param>
429    private void StartJobInAppDomain(Job myJob, JobData jobData) {
430      ClientCom.LogMessage("Received new job with id " + myJob.Id);
431      String pluginDir = Path.Combine(PluginCache.Instance.PluginTempBaseDir, myJob.Id.ToString());
432      bool pluginsPrepared = false;
433      string configFileName = string.Empty;
434
435      try {
436        PluginCache.Instance.PreparePlugins(myJob, out configFileName);
437        ClientCom.LogMessage("Plugins fetched for job " + myJob.Id);
438        pluginsPrepared = true;
439      }
440      catch (Exception exception) {
441        ClientCom.LogMessage(string.Format("Copying plugins for job {0} failed: {1}", myJob.Id, exception));
442      }
443
444      if (pluginsPrepared) {
445        try {
446          AppDomain appDomain = HeuristicLab.PluginInfrastructure.Sandboxing.SandboxManager.CreateAndInitSandbox(myJob.Id.ToString(), pluginDir, Path.Combine(pluginDir, configFileName));
447          appDomain.UnhandledException += new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
448          lock (engines) {
449            appDomains.Add(myJob.Id, appDomain);
450            ClientCom.LogMessage("Creating AppDomain");
451            Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
452            ClientCom.LogMessage("Created AppDomain");
453            engine.JobId = myJob.Id;
454            engine.Core = this;
455            ClientCom.LogMessage("Starting Engine for job " + myJob.Id);
456            engines.Add(myJob.Id, engine);
457            engine.Start(jobData.Data);
458            SlaveStatusInfo.JobsFetched++;
459            ClientCom.LogMessage("Increment FetchedJobs to:" + SlaveStatusInfo.JobsFetched);
460          }
461        }
462        catch (Exception exception) {
463          ClientCom.LogMessage("Creating the Appdomain and loading the job failed for job " + myJob.Id);
464          ClientCom.LogMessage("Error thrown is: " + exception.ToString());
465          KillAppDomain(myJob.Id);
466        }
467      }
468      heartbeatManager.AwakeHeartBeatThread();
469    }
470
471    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
472    private void OnExceptionOccured(Exception e) {
473      ClientCom.LogMessage("Error: " + e.ToString());
474      var handler = ExceptionOccured;
475      if (handler != null) handler(this, new EventArgs<Exception>(e));
476    }
477
478    private void AppDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
479      ClientCom.LogMessage("Exception in AppDomain: " + e.ExceptionObject.ToString());
480      KillAppDomain(new Guid(e.ExceptionObject.ToString()));
481    }
482
483    /// <summary>
484    /// Enqueues messages from the executor to the message queue.
485    /// This is necessary if the core thread has to execute certain actions, e.g.
486    /// killing of an app domain.
487    /// </summary>
488    /// <typeparam name="T"></typeparam>
489    /// <param name="action"></param>
490    /// <param name="parameter"></param>
491    /// <returns>true if the calling method can continue execution, else false</returns>
492    public void EnqueueExecutorMessage<T>(Action<T> action, T parameter) {
493      ExecutorMessageContainer<T> container = new ExecutorMessageContainer<T>();
494      container.Callback = action;
495      container.CallbackParameter = parameter;
496      MessageQueue.GetInstance().AddMessage(container);
497    }
498
499    /// <summary>
500    /// Kill a appdomain with a specific id.
501    /// </summary>
502    /// <param name="id">the GUID of the job</param>
503    //[MethodImpl(MethodImplOptions.Synchronized)]
504    public void KillAppDomain(Guid id) {
505      if (Thread.CurrentThread.ManagedThreadId != this.coreThreadId) {
506        EnqueueExecutorMessage<Guid>(KillAppDomain, id);
507        return;
508      }
509
510      ClientCom.LogMessage("Shutting down Appdomain for Job " + id);
511      lock (engines) {
512        try {
513          if (engines.ContainsKey(id)) {
514            engines[id].Dispose();
515            engines.Remove(id);
516          }
517
518          if (appDomains.ContainsKey(id)) {
519            appDomains[id].UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
520
521            int repeat = 5;
522            while (repeat > 0) {
523              try {
524                AppDomain.Unload(appDomains[id]);
525                repeat = 0;
526              }
527              catch (CannotUnloadAppDomainException) {
528                ClientCom.LogMessage("Could not unload AppDomain, will try again in 1 sec.");
529                Thread.Sleep(1000);
530                repeat--;
531                if (repeat == 0) {
532                  throw; // rethrow and let app crash
533                }
534              }
535            }
536            appDomains.Remove(id);
537          }
538
539          jobs.Remove(id);
540          PluginCache.Instance.DeletePluginsForJob(id);
541          GC.Collect();
542        }
543        catch (Exception ex) {
544          ClientCom.LogMessage("Exception when unloading the appdomain: " + ex.ToString());
545        }
546      }
547      GC.Collect();
548    }
549  }
550}
Note: See TracBrowser for help on using the repository browser.