Ntp.Common/Process/Scheduler.cs

//
// Copyright (c) 2013-2017 Carsten Sonne Larsen <cs@innolan.net>
//
// 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.Common.Log;

namespace Ntp.Common.Process
{
    /// <summary>
    /// A scheduler performs scheduling of jobs according to job schedule descriptions.
    /// </summary>
    public sealed class Scheduler : IScheduler, IDisposable
    {
        /// <summary>
        /// Initializes a new instance of the <see cref="Scheduler" /> class.
        /// </summary>
        /// <param name="log">Log.</param>
        public Scheduler(LogBase log)
        {
            StartTime = 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);

            Log = logGroup;
        }

        private readonly List<Job> jobs;
        private readonly List<Thread> runningThreads;
        private readonly List<ScheduledJob> schedule;
        private bool firstRun;

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

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

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

        /// <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 a value indicating whether this <see cref="Scheduler" /> is active.
        /// </summary>
        /// <value><c>true</c> if active; otherwise, <c>false</c>.</value>
        public bool Active { get; private set; }

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

        /// <summary>
        /// Gets the schedule.
        /// </summary>
        /// <value>The schedule.</value>
        public IEnumerable<ScheduledJob> Schedule => schedule;

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

        /// <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;

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

            var job = new Job(description, scheduleDescription, Log);
            jobs.Add(job);

            Log.SchedulerJobAdded(description, job);

            QueueJob(job, StartTime);
        }

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

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

                lock (schedule)
                {
                    Log.SchedulerStart(schedule.Count);
                }
            }

            // Take the next job of the queue
            ScheduledJob next;
            lock (schedule)
            {
                next = schedule.OrderBy(j => j.Run).ThenBy(j => j.Job.Description.Priority).First();
                schedule.Remove(next);
                next.Job.Queued = false;
                NextJob = next;
            }

            // Wait until job should be run
            int wait = Convert.ToInt32(next.Run.Subtract(DateTime.Now).TotalMilliseconds);
            if (wait < 0)
            {
                if (wait < -5000)
                    Log.SchedulerBehind();

                wait = 0;
            }

            bool signal = WaitHandle.WaitOne(wait);
            if (signal)
            {
                // Abort
                return;
            }

            // Check if the job is already running and postpone if needed
            lock (schedule)
            {
                if (next.Job.Description.ThreadType == ThreadType.SingleThreaded &&
                    schedule.Count(j => j.Job.Description.ThreadType == ThreadType.SingleThreaded && j.Job.Running) != 0)
                {
                    PostponeJob(next.Job);
                    return;
                }
            }

            Log.SchedulerJobExecuting(next);
            if (next.Job.Description.ThreadType != ThreadType.NoThread)
            {
                // Queue the job on the threadpool
                ThreadPool.QueueUserWorkItem(JobThreadStart, next.Job);
            }
            else
            {
                // Execute directly in scheduler thread
                ExecuteJob(next.Job);
            }

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

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

        public void Stop()
        {
            int count;

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

                count = runningThreads.Count;
            }

            if (count != 0)
                Log.SchedulerWaiting(count);

            lock (runningThreads)
            {
                foreach (var thread in runningThreads.ToList())
                {
                    // Wait maximum 15 seconds
                    thread.Join(15000/count);
                    if (!thread.IsAlive)
                        continue;

                    Log.SchedulerAbort(thread);
                    thread.Abort();
                }
            }

            Log.SchedulerFinished();
        }

        private void ExecuteJob(Job job)
        {
            try
            {
                job.Execute();
            }
            catch (Exception e)
            {
                Log.SchedulerError(job.Description.Name, e);
            }
        }

        private void JobThreadStart(object stateInfo)
        {
            try
            {
                lock (runningThreads)
                    runningThreads.Add(Thread.CurrentThread);

                var job = (Job) stateInfo;
                Thread.CurrentThread.CurrentCulture = CultureInfo.InvariantCulture;
                //Thread.CurrentThread.Name = job.Description.Name;
                job.Execute();
            }
            catch (Exception e)
            {
                Log.SchedulerError(Thread.CurrentThread.Name, e);
            }
            finally
            {
                lock (runningThreads)
                {
                    if (runningThreads.Contains(Thread.CurrentThread))
                    {
                        runningThreads.Remove(Thread.CurrentThread);
                    }
                }
            }
        }

        /// <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.SchedulerJobStatus(scheduledJob);
            }
        }

        /// <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)
            {
                lock (schedule)
                {
                    while (schedule.Count(j => Math.Abs(next.Subtract(j.Run).TotalMilliseconds) < 5000) != 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.SchedulerJobStatus(scheduledJob);
            }
        }

        #region IDisposable Support

        private bool disposedValue;

        private void Dispose(bool disposing)
        {
            if (disposedValue)
                return;

            if (disposing)
            {
                WaitHandle.Dispose();
            }

            disposedValue = true;
        }

        ~Scheduler()
        {
            Dispose(false);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion
    }
}