ntpa/Ntp.Process/Scheduler.cs

//
// Scheduler.cs
//
// Author:
//       Carsten Sonne Larsen <cs@innolan.dk>
//
// Copyright (c) 2013-2016 Carsten Sonne Larsen
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
using System;
using System.Collections;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using Ntp.Analyzer.Log;
using Ntp.Config;

namespace Ntp.Process
{
    /// <summary>
    /// A scheduler performs scheduling og jobs according to job schedule descriptions.
    /// </summary>
    public sealed class Scheduler : IScheduler
    {
        /// <summary>
        /// Initializes a new instance of the <see cref="Ntp.Process.Scheduler"/> class.
        /// </summary>
        /// <param name="log">Log.</param>
        public Scheduler (LogBase log)
        {
            start = DateTime.Now;
            firstRun = true;
            active = false;

            runningThreads = new List<Thread> ();
            waitHandle = new EventWaitHandle(false, EventResetMode.AutoReset);

            schedule = new List<ScheduledJob> ();
            jobs = new List<Job> ();
            activityLog = LogFactory.CreateActivityLog ();

            LogGroup logGroup = LogFactory.CreateGroupLog ();
            logGroup.Add (log);
            logGroup.Add (activityLog);

            this.log = logGroup;
        }

        private readonly List<Thread> runningThreads;
        private readonly EventWaitHandle waitHandle;
        private readonly ActivityLog activityLog;
        private readonly List<Job> jobs;
        private readonly List<ScheduledJob> schedule;
        private readonly LogBase log;
        private readonly DateTime start;
        private ScheduledJob nextJob;
        private bool firstRun;
        private bool active;

        /// <summary>
        /// Gets the enumerator.
        /// </summary>
        /// <returns>The enumerator.</returns>
        public IEnumerator<Job> GetEnumerator ()
        {
            return jobs.GetEnumerator ();
        }

        /// <summary>
        /// Gets the enumerator.
        /// </summary>
        /// <returns>The enumerator.</returns>
        IEnumerator IEnumerable.GetEnumerator ()
        {
            return jobs.GetEnumerator ();
        }

        /// <summary>
        /// Gets the wait handle of this <see cref="Scheduler" />.
        /// </summary>
        /// <value>The wait handle.</value>
        public EventWaitHandle WaitHandle {
            get { return waitHandle; }
        }

        /// <summary>
        /// Gets the startup time of this <see cref="Scheduler" />.
        /// </summary>
        /// <value>The startup time.</value>
        public DateTime StartTime {
            get { return start; }
        }

        /// <summary>
        /// Gets a value indicating whether this <see cref="Ntp.Process.Scheduler"/> is active.
        /// </summary>
        /// <value><c>true</c> if active; otherwise, <c>false</c>.</value>
        public bool Active {
            get { return active; }
        }

        /// <summary>
        /// Gets the log used by this <see cref="Scheduler" />.
        /// </summary>
        /// <value>The log.</value>
        public LogBase Log {
            get { return log; }
        }

        /// <summary>
        /// Gets the activity log.
        /// </summary>
        /// <value>The activity log.</value>
        public ActivityLog ActivityLog {
            get { return activityLog; }
        }

        /// <summary>
        /// Gets the schedule.
        /// </summary>
        /// <value>The schedule.</value>
        public IEnumerable<ScheduledJob> Schedule {
            get { return schedule; }
        }

        /// <summary>
        /// Gets the next job to be executed.
        /// </summary>
        /// <value>The next job.</value>
        public ScheduledJob NextJob {
            get { return nextJob; }
        }

        /// <summary>
        /// Add the specified job to the scheduler queue.
        /// </summary>
        /// <param name="description">Description.</param>
        public void Add (JobDescription description)
        {
            if (description.Configuration == null || description.Configuration.Frequency == -1)
                return;

            JobScheduleDescription schedule = new JobScheduleDescription (
                                                  description.Configuration.InitialRun,
                                                  description.Configuration.FixedRun,
                                                  description.Configuration.Frequency);

            Job job = new Job (description, schedule, log);

            jobs.Add (job);
            QueueJob (job, start);

            log.WriteLine ("Job added to scheduler: " + job.ConfigString, Severity.Debug);
        }

        /// <summary>
        /// Run the scheduler queue pump method.
        /// </summary>
        public void RunOneCycle ()
        {
            Thread.CurrentThread.CurrentCulture = CultureInfo.InvariantCulture;

            if (firstRun) {
                firstRun = false;
                active = true;

                log.WriteLine (String.Format ("Starting scheduler with {0} jobs.", schedule.Count ()), Severity.Notice);
            }

            lock (runningThreads) {
                foreach (Thread finished in runningThreads.Where (t => !t.IsAlive).ToList())
                    runningThreads.Remove (finished);
            }

            ScheduledJob next = schedule.OrderBy (j => j.Run).ThenBy (j => j.Job.Description.Priority).First ();
            schedule.Remove (next);
            next.Job.Queued = false;
            nextJob = next;

            Thread thread = new Thread (new ThreadStart (next.Job.Execute));
            thread.CurrentCulture = CultureInfo.InvariantCulture;
            thread.Name = next.Job.Description.Name;

            int wait = Convert.ToInt32 (next.Run.Subtract (DateTime.Now).TotalMilliseconds);

            if (wait < 0) {
                if (wait < -5000)
                    log.WriteLine ("Behind schedule. Trying to catch up.", Severity.Info);

                wait = 0;
            }

            bool signal = waitHandle.WaitOne (wait);
            if (signal) {
                QueueJob (next.Job, DateTime.Now.Subtract (TimeSpan.FromMilliseconds (wait)));
                return;
            }

            if (next.Job.Description.SingleThread && schedule.Count (j => j.Job.Description.SingleThread && j.Job.Running) != 0) {
                PostponeJob (next.Job);
                return;
            }

            lock (runningThreads) {
                runningThreads.Add (thread);
                thread.Start ();
            }

            // Dont re-schedule "run-only-once" jobs.
            if (next.Job.Schedule.Frequency == 0)
                return;

            QueueJob (next.Job, DateTime.Now);
        }

        /// <summary>
        /// Queue job for scheduled run.
        /// </summary>
        /// <param name="job">Job.</param>
        /// <param name="run">Run.</param>
        private void QueueJob (Job job, DateTime run)
        {
            DateTime next = job.Schedule.CalculateNextRun (run);
            double offset = 0;

            // Adjust offset to avoid simultaneously job execution.
            if (job.Schedule.CanMove) {
                while (schedule.Where (j => Math.Abs (next.Subtract (j.Run).TotalMilliseconds) < 5000).Count () != 0) {
                    double move = job.Schedule.Frequency / 40.0;
                    next = next.AddMinutes (move);
                    offset += move;
                }
            }

            lock (schedule) {
                ScheduledJob scheduledJob = job.Schedule.CreateNew (job, run, offset);
                schedule.Add (scheduledJob);
                job.Queued = true;
                job.Postponed = false;
                log.WriteLine (scheduledJob.ToString (), Severity.Debug);
            }
        }

        /// <summary>
        /// Postpones the job when putting it to the queue.
        /// </summary>
        /// <param name="job">Job.</param>
        private void PostponeJob (Job job)
        {
            lock (schedule) {
                ScheduledJob scheduledJob = job.Schedule.CreatePostponed (job);
                schedule.Add (scheduledJob);
                job.Queued = true;
                job.Postponed = true;
                log.WriteLine (scheduledJob.ToString (), Severity.Debug);
            } 
        }

        public void Stop()
        {
            int count;

            lock (runningThreads) {
                foreach (Thread finished in runningThreads.Where (t => !t.IsAlive).ToList())
                    runningThreads.Remove (finished);

                count = runningThreads.Count;
            }

            if (count != 0)
                log.WriteLine (String.Format (
                    "Waiting for {0} threads to finish.", count),
                    Severity.Notice);

            lock (runningThreads) {
                foreach (Thread thread in runningThreads.ToList()) {
                    thread.Join (30000 / count);
                    if (thread.IsAlive) {
                        log.WriteLine (String.Format (
                            "Arborting thread {0}.", thread.Name),
                            Severity.Warn);
                        thread.Abort ();
                    }
                }
            }

            log.WriteLine (String.Format (
                "All threads finished.", runningThreads.Count),
                Severity.Notice);
        }
    }
}