//******************************************************************************************************
//  LogicalThreadScheduler.cs - Gbtc
//
//  Copyright © 2015, Grid Protection Alliance.  All Rights Reserved.
//
//  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
//  the NOTICE file distributed with this work for additional information regarding copyright ownership.
//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may
//  not use this file except in compliance with the License. You may obtain a copy of the License at:
//
//      http://opensource.org/licenses/MIT
//
//  Unless agreed to in writing, the subject software distributed under the License is distributed on an
//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
//  License for the specific language governing permissions and limitations.
//
//  Code Modification History:
//  ----------------------------------------------------------------------------------------------------
//  09/23/2015 - Stephen C. Wills
//       Generated original version of source code.
//
//******************************************************************************************************

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;

namespace GSF.Threading
{
    /// <summary>
    /// Manages synchronization of actions by dispatching actions
    /// to logical threads to be processed synchronously.
    /// </summary>
    public class LogicalThreadScheduler
    {
        #region [ Members ]

        // Events

        /// <summary>
        /// Triggered when an action managed by this
        /// synchronization manager throws an exception.
        /// </summary>
        public event EventHandler<EventArgs<Exception>> UnhandledException;

        // Fields
        private ConcurrentQueue<Func<LogicalThread>>[] m_logicalThreadQueues;
        private int m_maxThreadCount;
        private int m_threadCount;
        private bool m_useBackgroundThreads;

        #endregion

        #region [ Constructors ]

        /// <summary>
        /// Creates a new instance of the <see cref="LogicalThreadScheduler"/> class.
        /// </summary>
        /// <param name="priorityLevels">The number of levels of priority supported by this logical thread scheduler.</param>
        /// <exception cref="ArgumentException"><paramref name="priorityLevels"/> is less than or equal to zero.</exception>
        public LogicalThreadScheduler(int priorityLevels = 1)
        {
            if (priorityLevels < 1)
                throw new ArgumentException($"A logical thread scheduler must have at least one priority level.", nameof(priorityLevels));

            m_maxThreadCount = Environment.ProcessorCount;
            m_logicalThreadQueues = new ConcurrentQueue<Func<LogicalThread>>[priorityLevels];
            m_useBackgroundThreads = true;

            for (int i = 0; i < priorityLevels; i++)
                m_logicalThreadQueues[i] = new ConcurrentQueue<Func<LogicalThread>>();
        }

        #endregion

        #region [ Properties ]

        /// <summary>
        /// Gets or sets the target for the maximum number of physical
        /// threads managed by this synchronization manager at any given time.
        /// </summary>
        public int MaxThreadCount
        {
            get
            {
                return Interlocked.CompareExchange(ref m_maxThreadCount, 0, 0);
            }
            set
            {
                if (value <= 0)
                    throw new ArgumentException("Max thread count must be greater than zero", nameof(value));

                int diff = value - m_maxThreadCount;
                int inactiveThreads = Math.Min(diff, m_logicalThreadQueues.Sum(queue => queue.Count));

                Interlocked.Exchange(ref m_maxThreadCount, value);

                for (int i = 0; i < inactiveThreads; i++)
                    ActivatePhysicalThread();
            }
        }

        /// <summary>
        /// Gets or sets the flag that determines whether the threads in
        /// the scheduler's thread pool should be background threads.
        /// </summary>
        public bool UseBackgroundThreads
        {
            get
            {
                return m_useBackgroundThreads;
            }
            set
            {
                m_useBackgroundThreads = value;
            }
        }

        /// <summary>
        /// Gets the number of levels of priority
        /// supported by this scheduler.
        /// </summary>
        public int PriorityLevels
        {
            get
            {
                return m_logicalThreadQueues.Length;
            }
        }

        /// <summary>
        /// Gets the current number of active physical threads.
        /// </summary>
        private int ThreadCount
        {
            get
            {
                return Interlocked.CompareExchange(ref m_threadCount, 0, 0);
            }
        }

        #endregion

        #region [ Methods ]

        /// <summary>
        /// Creates a new logical thread whose
        /// execution is managed by this scheduler.
        /// </summary>
        /// <returns>A new logical thread managed by this scheduler.</returns>
        public LogicalThread CreateThread()
        {
            return new LogicalThread(this);
        }

        /// <summary>
        /// Signals the manager when a logical
        /// thread has new actions to be processed.
        /// </summary>
        /// <param name="thread">The thread with new actions to be processed.</param>
        /// <param name="priority">The priority at which the thread is being signaled.</param>
        internal void SignalItemHandler(LogicalThread thread, int priority)
        {
            if (thread.TryActivate(priority))
            {
                // Make sure the thread has an action to be processed because
                // another thread could have processed all remaining actions and
                // deactivated the thread right before the call to TryActivate()
                if (!thread.HasAction)
                {
                    thread.Deactivate();
                    return;
                }

                Enqueue(thread);
                ActivatePhysicalThread();
            }
        }

        /// <summary>
        /// Activates a new physical thread if the thread
        /// count has not yet reached its maximum limit.
        /// </summary>
        private void ActivatePhysicalThread()
        {
            int threadCount;
            int newThreadCount;

            threadCount = ThreadCount;

            while (threadCount < Interlocked.CompareExchange(ref m_maxThreadCount, 0, 0))
            {
                newThreadCount = Interlocked.CompareExchange(ref m_threadCount, threadCount + 1, threadCount);

                if (newThreadCount == threadCount)
                {
                    StartNewPhysicalThread();
                    break;
                }

                threadCount = newThreadCount;
            }
        }

        /// <summary>
        /// Starts a new physical thread to
        /// process actions from logical threads.
        /// </summary>
        private void StartNewPhysicalThread()
        {
            Thread thread = new Thread(ProcessLogicalThreads);
            thread.IsBackground = m_useBackgroundThreads;
            thread.Start();
        }

        /// <summary>
        /// Processes the next available task from the least recently
        /// processed logical thread that is available for processing.
        /// </summary>
        private void ProcessLogicalThreads()
        {
            Stopwatch stopwatch;
            Func<LogicalThread> accessor = null;
            LogicalThread thread;
            Action action;

            stopwatch = new Stopwatch();

            while (ThreadCount <= MaxThreadCount && (object)m_logicalThreadQueues.FirstOrDefault(queue => queue.TryDequeue(out accessor)) != null)
            {
                thread = accessor();

                if ((object)thread == null)
                    continue;

                action = thread.Pull();

                if ((object)action != null)
                {
                    LogicalThread.CurrentThread = thread;

                    stopwatch.Restart();
                    TryExecute(action);
                    stopwatch.Stop();

                    LogicalThread.CurrentThread = null;
                    thread.UpdateStatistics(stopwatch.Elapsed);
                }

                if (thread.HasAction)
                    Enqueue(thread);
                else
                    thread.Deactivate();
            }

            DeactivatePhysicalThread();

            if (!m_logicalThreadQueues.All(queue => queue.IsEmpty))
                ActivatePhysicalThread();
        }

        /// <summary>
        /// Queues the given thread for execution.
        /// </summary>
        /// <param name="thread">The thread to be queued for execution.</param>
        private void Enqueue(LogicalThread thread)
        {
            ICancellationToken executionToken;
            int activePriority;
            int nextPriority;

            do
            {
                // Create the execution token to be used in the closure
                ICancellationToken nextExecutionToken = new CancellationToken();

                // Always update the thread's active priority before
                // the execution token to mitigate race conditions
                nextPriority = thread.NextPriority;
                thread.ActivePriority = nextPriority;
                thread.NextExecutionToken = nextExecutionToken;

                // Now that the action can be cancelled by another thread using the
                // new cancellation token, it should be safe to put it in the queue
                m_logicalThreadQueues[PriorityLevels - nextPriority].Enqueue(() =>
                {
                    if (nextExecutionToken.Cancel())
                        return thread;

                    return null;
                });

                // Because enqueuing the thread is a multi-step process, we need to
                // double-check in case the thread's priority changed in the meantime
                activePriority = thread.ActivePriority;
                nextPriority = thread.NextPriority;

                // We can use the cancellation token we just created because we only
                // really need to double-check the work that was done on this thread;
                // in other words, if another thread changed the priority in the
                // meantime, it can double-check its own work
                executionToken = nextExecutionToken;
            }
            while (activePriority != nextPriority && executionToken.Cancel());
        }

        /// <summary>
        /// Attempts to execute the given action.
        /// </summary>
        /// <param name="action">The action to be executed.</param>
        private void TryExecute(Action action)
        {
            try
            {
                action();
            }
            catch (Exception ex)
            {
                if (!TryHandleException(ex))
                    throw;
            }
        }

        /// <summary>
        /// Decrements the thread count upon deactivation of a physical thread.
        /// </summary>
        private void DeactivatePhysicalThread()
        {
            Interlocked.Decrement(ref m_threadCount);
        }

        /// <summary>
        /// Attempts to handle the exception via either the logical thread's
        /// exception handler or the scheduler's exception handler.
        /// </summary>
        /// <param name="unhandledException">The unhandled exception thrown by an action on the logical thread.</param>
        /// <returns>True if the exception could be handled; false otherwise.</returns>
        private bool TryHandleException(Exception unhandledException)
        {
            AggregateException aggregateException;
            StringBuilder message;
            bool handled;

            message = new StringBuilder();
            message.AppendFormat("Logical thread action threw an exception of type {0}: {1}", unhandledException.GetType().FullName, unhandledException.Message);
            aggregateException = new AggregateException(message.ToString(), unhandledException);

            try
            {
                // Attempt to handle the exception via the logical thread's exception handler
                handled = LogicalThread.CurrentThread.OnUnhandledException(unhandledException);
            }
            catch (Exception handlerException)
            {
                // If the handler throws an exception,
                // make a note of it in the exception's exception message
                message.AppendLine();
                message.AppendFormat("Logical thread exception handler threw an exception of type {0}: {1}", handlerException.GetType().FullName, handlerException.Message);
                aggregateException = new AggregateException(message.ToString(), aggregateException.InnerExceptions.Concat(new Exception[] { handlerException }));
                handled = false;
            }

            try
            {
                // If the logical thread's exception handler was not able to handle the exception,
                // attempt to handle the exception via the thread scheduler's exception handler
                Exception ex = (aggregateException.InnerExceptions.Count > 1) ? aggregateException : unhandledException;
                handled = handled || OnUnhandledException(ex);
            }
            catch (Exception handlerException)
            {
                // If the handler throws an exception,
                // make a note of it in the exception's exception message
                message.AppendLine();
                message.AppendFormat("Scheduler exception handler threw an exception of type {0}: {1}", handlerException.GetType().FullName, handlerException.Message);
                aggregateException = new AggregateException(message.ToString(), aggregateException.InnerExceptions.Concat(new Exception[] { handlerException }));
                handled = false;
            }

            if (!handled)
            {
                // If the exception could not be handled by either the
                // logical thread's exception handler or the scheduler's
                // exception handler, throw it as an unhandled exception
                if (aggregateException.InnerExceptions.Count > 1)
                    throw aggregateException;

                return false;
            }

            return true;
        }

        /// <summary>
        /// Raises the <see cref="UnhandledException"/> event.
        /// </summary>
        /// <param name="ex">The unhandled exception.</param>
        /// <returns>True if there are any handlers attached to this event; false otherwise.</returns>
        private bool OnUnhandledException(Exception ex)
        {
            EventHandler<EventArgs<Exception>> unhandledException = UnhandledException;

            if ((object)unhandledException == null)
                return false;

            unhandledException(this, new EventArgs<Exception>(ex));

            return true;
        }

        #endregion
    }
}