source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/HeartbeatManager.cs @ 5728

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2010 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */
#endregion

using System;
using System.Collections.Generic;
using System.Threading;
using HeuristicLab.Common;

namespace HeuristicLab.Clients.Hive.SlaveCore {
  /// <summary>
  /// Heartbeat class. It sends every x ms a heartbeat to the server and receives a Message
  /// </summary>
  public class HeartbeatManager {
    private static object locker = new object();
    public TimeSpan Interval { get; set; }
    private Thread heartBeatThread;
    private AutoResetEvent waitHandle;
    private WcfService wcfService;
    private bool threadStopped;
    ReaderWriterLockSlim heartBeatThreadIsSleepingLock = new ReaderWriterLockSlim();

    public HeartbeatManager() {
      Interval = new TimeSpan(0, 0, 10);
    }

    public HeartbeatManager(TimeSpan interval) {
      Interval = interval;
    }

    /// <summary>
    /// Starts the Heartbeat signal.
    /// </summary>
    public void StartHeartbeat() {
      this.waitHandle = new AutoResetEvent(true);
      wcfService = WcfService.Instance;
      threadStopped = false;
      heartBeatThread = new Thread(RunHeartBeatThread);
      heartBeatThread.Start();
    }

    /// <summary>
    /// Stop the heartbeat
    /// </summary>
    public void StopHeartBeat() {
      threadStopped = true;
      waitHandle.Set();
      heartBeatThread.Join();
    }

    /// <summary>
    /// use this method to singalize there is work to do (to avoid the waiting period if its clear that actions are required)
    /// </summary>
    public void AwakeHeartBeatThread() {
      if (!threadStopped)
        waitHandle.Set();
    }

    private void RunHeartBeatThread() {
      bool sleepForever;
      while (!threadStopped) {
        sleepForever = false;
        try {
          lock (locker) {
            if (wcfService.ConnState != NetworkEnum.WcfConnState.Connected) {
              wcfService.Connect(ConfigManager.Instance.GetClientInfo()); // Login happens automatically upon successfull connection
            }
            if (wcfService.ConnState == NetworkEnum.WcfConnState.Connected) {
              Slave info = ConfigManager.Instance.GetClientInfo();

              Heartbeat heartBeatData = new Heartbeat {
                SlaveId = info.Id,
                FreeCores = info.Cores.HasValue ? info.Cores.Value - ConfigManager.Instance.GetUsedCores() : 0,
                FreeMemory = ConfigManager.GetFreeMemory(),
                JobProgress = ConfigManager.Instance.GetExecutionTimeOfAllJobs(),
                AssignJob = true //TODO: check if we want another job
              };

              SlaveClientCom.Instance.ClientCom.LogMessage("Sending Heartbeat: " + heartBeatData);
              List<MessageContainer> msgs = wcfService.SendHeartbeat(heartBeatData);

              if (msgs == null) {
                SlaveClientCom.Instance.ClientCom.LogMessage("Error getting response from Heartbeat");
                OnExceptionOccured(new Exception("Error getting response from Heartbeat"));
              } else {
                SlaveClientCom.Instance.ClientCom.LogMessage("Heartbeat Response received (" + msgs.Count + "): ");
                msgs.ForEach(mc => SlaveClientCom.Instance.ClientCom.LogMessage(mc.Message.ToString()));
                msgs.ForEach(mc => MessageQueue.GetInstance().AddMessage(mc));
                //after fetching a job, we sleep until the core wakes us up!!
                msgs.ForEach(s => { if (s.Message == MessageContainer.MessageType.CalculateJob) sleepForever = true; });
              }
            }
          }
        }
        catch (Exception e) {
          SlaveClientCom.Instance.ClientCom.LogMessage("Heartbeat Thread failed badly: " + e.Message);
          OnExceptionOccured(e);
        }
        if (sleepForever)
          waitHandle.WaitOne();
        else
          waitHandle.WaitOne(this.Interval);
      }
      waitHandle.Close();
      SlaveClientCom.Instance.ClientCom.LogMessage("Heartbeat thread stopped");
    }


    #region Eventhandler
    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
    private void OnExceptionOccured(Exception e) {
      var handler = ExceptionOccured;
      if (handler != null) handler(this, new EventArgs<Exception>(e));
    }
    #endregion
  }
}