//******************************************************************************************************
//  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<LogicalThread> m_logicalThreads;
        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>
        public LogicalThreadScheduler()
        {
            m_maxThreadCount = Environment.ProcessorCount;
            m_logicalThreads = new ConcurrentQueue<LogicalThread>();
            m_useBackgroundThreads = true;
        }

        #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
            {
                int diff = value - m_maxThreadCount;
                int inactiveThreads = Math.Min(diff, m_logicalThreads.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 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 CreateThread(1);
        }

        /// <summary>
        /// Creates a new logical thread whose
        /// execution is managed by this scheduler.
        /// </summary>
        /// <param name="priorityLevels">The number of levels of priority supported by the logical thread.</param>
        /// <returns>A new logical thread managed by this scheduler.</returns>
        public LogicalThread CreateThread(int priorityLevels)
        {
            return new LogicalThread(this, priorityLevels);
        }

        /// <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>
        internal void SignalItemHandler(LogicalThread thread)
        {
            if (thread.TryActivate())
            {
                m_logicalThreads.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;
            LogicalThread thread;
            Action action;

            stopwatch = new Stopwatch();

            while (ThreadCount <= MaxThreadCount && m_logicalThreads.TryDequeue(out thread))
            {
                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)
                    m_logicalThreads.Enqueue(thread);
                else
                    thread.Deactivate();
            }

            DeactivatePhysicalThread();

            if (!m_logicalThreads.IsEmpty)
                ActivatePhysicalThread();
        }

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