source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/ConsoleTests/Program.cs @ 6451

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using HeuristicLab.Algorithms.GeneticAlgorithm;
using HeuristicLab.Clients.Hive.Jobs;
using HeuristicLab.Clients.Hive.SlaveCore;
using HeuristicLab.Common;
using HeuristicLab.Core;
using HeuristicLab.Hive;
using HeuristicLab.Persistence.Default.Xml;
using HeuristicLab.PluginInfrastructure;
using HeuristicLab.Problems.TestFunctions;

namespace HeuristicLab.Clients.Hive.Slave.Tests {
  #region Test Setup
  public class Program {
    static void Main(string[] args) {
      //TestThreadSafeLog(); return;
      //TestThreadSafePersistence(); return;
      TestCappedLog(); return;
      string path = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);

      DeleteDirectory(Path.Combine(path, "PluginCache"));
      DeleteDirectory(Path.Combine(path, "PluginTemp"));

      var pm = new HeuristicLab.PluginInfrastructure.Manager.PluginManager(path);
      pm.DiscoverAndCheckPlugins();
      var app = pm.Applications.Where(x => x.Name == "HeuristicLab.Clients.Hive.Slave.Tests").Single();
      pm.Run(app);
    }

    private static void TestCappedLog() {
      ILog log = new Log(); // maxMessageCount = -1
      Console.WriteLine("0=={0}", log.Messages.Count());
      log.LogMessage("asdf");
      Console.WriteLine("1=={0}", log.Messages.Count());
      log.LogMessage("asdf");
      log.LogMessage("asdf");
      Console.WriteLine("3=={0}", log.Messages.Count());
      
      ILog log0 = new Log(0);
      Console.WriteLine("0=={0}", log0.Messages.Count());
      log0.LogMessage("asdf");
      Console.WriteLine("0=={0}", log0.Messages.Count());
      log0.LogMessage("asdf");
      log0.LogMessage("asdf");
      log0.LogMessage("asdf");
      Console.WriteLine("0=={0}", log0.Messages.Count());

      ILog log1 = new Log(1);
      log1.LogMessage("asdf 1");
      Console.WriteLine("1=={0}", log1.Messages.Count());
      log1.LogMessage("asdf 2");
      log1.LogMessage("asdf 3");
      Console.WriteLine("1=={0}", log1.Messages.Count());
      Console.WriteLine("asdf 3=={0}", log1.Messages.First());
    }

    private static void TestThreadSafeLog() {
      ILog log = new ThreadSafeLog(new Log());
      bool abortOutputThread = false;

      Task outputTask = Task.Factory.StartNew(() => {
        while (!abortOutputThread) {
          Console.Clear();
          foreach (var m in log.Messages) {
            Console.WriteLine(m);
          }
        }
      });

      int n = 50;
      Task[] tasks = new Task[n];
      for (int i = 0; i < n; i++) {
        tasks[i] = Task.Factory.StartNew((idx) => {
          for (int j = 0; j < 5000; j++) {
            log.LogMessage(string.Format("Message {0}, Task {1}", j, idx));
          }
        }, i);
      }
      Task.WaitAll(tasks);
      abortOutputThread = true;
      outputTask.Wait();
      Console.WriteLine("Finished, {0} messages", log.Messages.Count());
      Console.ReadLine();
    }

    private static void TestThreadSafePersistence() {
      int n = 100;
      Task[] tasks = new Task[n];
      for (int i = 0; i < n; i++) {
        tasks[i] = Task.Factory.StartNew((idx) => {
          var alg = new GeneticAlgorithm { Engine = new SequentialEngine.SequentialEngine(), Problem = new SingleObjectiveTestFunctionProblem() };
          byte[] data = Serialize(alg);
          Console.WriteLine("Object #{0} serialized", idx);
        }, i);
      }
      Task.WaitAll(tasks);
    }
    public static byte[] Serialize(object obj) {
      MemoryStream memStream = new MemoryStream();
      XmlGenerator.Serialize(obj, memStream);
      byte[] jobByteArray = memStream.ToArray();
      memStream.Dispose();
      return jobByteArray;
    }

    public static bool DeleteDirectory(string target_dir) {
      if (!Directory.Exists(target_dir))
        return true;

      bool result = false;

      string[] files = Directory.GetFiles(target_dir);
      string[] dirs = Directory.GetDirectories(target_dir);

      foreach (string file in files) {
        File.SetAttributes(file, FileAttributes.Normal);
        File.Delete(file);
      }

      foreach (string dir in dirs) {
        DeleteDirectory(dir);
      }

      Directory.Delete(target_dir, false);

      return result;
    }
  }
  #endregion

  [Plugin("HeuristicLab.Clients.Hive.Slave.Tests", "1.0.0.0")]
  [PluginFile("HeuristicLab.Clients.Hive.Slave-3.4.Tests.exe", PluginFileType.Assembly)]
  public class TestPlugin : PluginBase { }

  [Application("HeuristicLab.Clients.Hive.Slave.Tests")]
  public class TestApp : ApplicationBase {

    public override void Run() {
      var mockPluginService = new MockPluginService();
      var pluginManager = new PluginManager(mockPluginService, new Log());
      JobManager jobManager = new JobManager(pluginManager, new Log());

      jobManager.JobStarted += new EventHandler<EventArgs<SlaveJob>>(jobManager_JobStarted);
      jobManager.JobPaused += new EventHandler<EventArgs<SlaveJob, JobData>>(jobManager_JobPaused);
      jobManager.JobStopped += new EventHandler<EventArgs<SlaveJob, JobData>>(jobManager_JobStopped);
      jobManager.JobFailed += new EventHandler<EventArgs<Tuple<SlaveJob, JobData, Exception>>>(jobManager_JobFailed);
      jobManager.ExceptionOccured += new EventHandler<EventArgs<SlaveJob, Exception>>(jobManager_ExceptionOccured);

      ConfigManager.Instance = new ConfigManager(jobManager);

      //TestCalculateJobs(mockPluginService, jobManager);
      //TestPauseJob(mockPluginService, jobManager);
      //TestStopJob(mockPluginService, jobManager);
      //TestFailJob(mockPluginService, jobManager);

      //TestAbortAll(mockPluginService, jobManager);
      //TestStopAll(mockPluginService, jobManager);
      //TestPauseAll(mockPluginService, jobManager);

      Console.WriteLine(jobManager.JobCount);
    }

    // calculate many jobs and request execution times rapidly
    private static void TestCalculateJobs(MockPluginService mockPluginService, JobManager jobManager) {
      StartStatusObservationThread();

      Task[] tasks = CreateAndStartJobs(mockPluginService, jobManager, 10, 1000);
      Task.WaitAll(tasks);

      Console.ReadLine();
    }

    // calculate jobs and stop them
    private void TestStopJob(MockPluginService mockPluginService, JobManager jobManager) {
      for (int i = 0; i < 5; i++) {
        var optimizerJob = new OptimizerJob(new GeneticAlgorithm { Engine = new SequentialEngine.SequentialEngine(), Problem = new SingleObjectiveTestFunctionProblem() });

        Job job;
        JobData jobData;
        CreateJob(mockPluginService, optimizerJob, out job, out jobData);

        jobManager.StartJobAsync(job, jobData);
        Thread.Sleep(i * 1000);

        jobManager.StopJobAsync(job.Id);
      }
    }

    // calculate jobs and pause them
    private void TestPauseJob(MockPluginService mockPluginService, JobManager jobManager) {
      for (int i = 4; i < 5; i++) {
        var optimizerJob = new OptimizerJob(new GeneticAlgorithm { Engine = new SequentialEngine.SequentialEngine(), Problem = new SingleObjectiveTestFunctionProblem() });

        Job job;
        JobData jobData;
        CreateJob(mockPluginService, optimizerJob, out job, out jobData);

        jobManager.StartJobAsync(job, jobData);
        Thread.Sleep(i * 1000);

        jobManager.PauseJobAsync(job.Id);
      }
    }

    // start many jobs and then abort all
    private void TestAbortAll(MockPluginService mockPluginService, JobManager jobManager) {
      StartStatusObservationThread();

      Task[] tasks = CreateAndStartJobs(mockPluginService, jobManager, 4, 5000);
      Task.WaitAll(tasks);

      Console.WriteLine("Aborting all jobs");
      jobManager.AbortAllJobs();
      Console.ReadLine();
    }

    // start many jobs and then stop all and observe results from GA
    private void TestStopAll(MockPluginService mockPluginService, JobManager jobManager) {
      StartStatusObservationThread();

      Task[] tasks = CreateAndStartJobs(mockPluginService, jobManager, 8, 5000);
      Task.WaitAll(tasks);

      Console.WriteLine("Stopping all jobs");
      jobManager.StopAllJobsAsync();
      Console.ReadLine();
    }

    // start many jobs and then pause all and observe results from GA
    private void TestPauseAll(MockPluginService mockPluginService, JobManager jobManager) {
      StartStatusObservationThread();

      Task[] tasks = CreateAndStartJobs(mockPluginService, jobManager, 4, 1000);
      Task.WaitAll(tasks);

      Console.WriteLine("Pausing all jobs");
      jobManager.PauseAllJobsAsync();
      Console.ReadLine();
    }

    #region Helpers
    private static Task[] CreateAndStartJobs(MockPluginService mockPluginService, JobManager jobManager, int n, int generations = 1000) {
      Task[] tasks = new Task[n];
      for (int i = 0; i < n; i++) {
        tasks[i] = Task.Factory.StartNew((idx) => {
          var ga = new GeneticAlgorithm { Engine = new SequentialEngine.SequentialEngine(), Problem = new SingleObjectiveTestFunctionProblem() };
          ga.MaximumGenerations.Value = generations;
          var mockJob = new OptimizerJob(ga);

          Job job;
          JobData jobData;
          CreateJob(mockPluginService, mockJob, out job, out jobData);

          jobManager.StartJobAsync(job, jobData);
        }, i);
      }
      return tasks;
    }

    private static void StartStatusObservationThread(int intervalMs = 5000) {
      Task.Factory.StartNew(() => {
        while (true) {
          Thread.Sleep(intervalMs);
          PrintInfo();
        }
      });
    }

    private static void PrintInfo() {
      var info = ConfigManager.Instance.GetClientInfo();
      var heartBeatData = new Heartbeat {
        SlaveId = info.Id,
        FreeCores = info.Cores.HasValue ? info.Cores.Value - SlaveStatusInfo.UsedCores : 0,
        FreeMemory = ConfigManager.GetFreeMemory(),
        CpuUtilization = ConfigManager.Instance.GetCpuUtilization(),
        JobProgress = ConfigManager.Instance.GetExecutionTimeOfAllJobs(),
        AssignJob = !ConfigManager.Instance.Asleep
      };
      Console.WriteLine(heartBeatData);
      Console.WriteLine(ConfigManager.Instance.GetStatusForClientConsole().ToString());
    }

    private static object locker = new object();
    private static void CreateJob(MockPluginService mockPluginService, IJob mockJob, out Job job, out JobData jobData) {
      job = new Job { Id = Guid.NewGuid(), CoresNeeded = 1, MemoryNeeded = 0 };
      job.PluginsNeededIds = new List<Guid>();
      jobData = new JobData { JobId = job.Id };
      IEnumerable<Type> types;
      jobData.Data = PersistenceUtil.Serialize(mockJob, out types);

      // make plugins available through mockPluginService
      var plugins = new List<IPluginDescription>();
      PluginUtil.CollectDeclaringPlugins(plugins, types);
      foreach (var plugin in plugins) {
        var p = PluginUtil.CreatePlugin(plugin);
        p.Id = Guid.NewGuid();
        lock (locker) {
          if (!mockPluginService.Plugins.ContainsKey(p.Id)) {
            mockPluginService.Plugins[p.Id] = p;
            mockPluginService.PluginDatas[p.Id] = PluginUtil.CreatePluginDatas(plugin);
          }
        }
        foreach (var pd in mockPluginService.PluginDatas[p.Id]) {
          pd.Id = Guid.NewGuid();
          pd.PluginId = p.Id;
        }
        job.PluginsNeededIds.Add(p.Id);
      }
    }
    #endregion

    #region Events
    private void jobManager_JobStarted(object sender, EventArgs<SlaveJob> e) {
      Console.WriteLine("# Job Started: {0}", e.Value.JobId);
    }
    private void jobManager_JobPaused(object sender, EventArgs<SlaveJob, JobData> e) {
      Console.WriteLine("# Job Paused: {0}", e.Value.JobId);
      PrintJobResultInfo(e.Value2);
    }

    private void jobManager_JobStopped(object sender, EventArgs<SlaveJob, JobData> e) {
      Console.WriteLine("# Job Stopped: {0}", e.Value.JobId);
      PrintJobResultInfo(e.Value2);
    }
    private void jobManager_JobFailed(object sender, EventArgs<Tuple<SlaveJob, JobData, Exception>> e) {
      Console.WriteLine("# Job Failed: {0}", e.Value.Item3.ToString());
      if (e.Value.Item2 != null) {
        PrintJobResultInfo(e.Value.Item2);
      }
    }
    private void jobManager_ExceptionOccured(object sender, EventArgs<SlaveJob, Exception> e) {
      Console.WriteLine("# Exception Occured: {0}", e.Value2);
    }

    private static void PrintJobResultInfo(JobData jobData) {
      var ga = (GeneticAlgorithm)PersistenceUtil.Deserialize<OptimizerJob>(jobData.Data).Item;
      var results = ga.Results;
      Console.WriteLine("# State: {0}, ExecTime: {1}, Generations: {2}", ga.ExecutionState, ga.ExecutionTime, results.ContainsKey("Generations") ? results["Generations"].Value.ToString() : "-");
    }
    #endregion
  }
}