GSF.Core: Updated LogicalThread API to handle prioritization at the thread scheduler so that threads can be reprioritized based on the priority of their tasks.

StephenCWills committed 9 years ago

Commit d6482bb1 parent 3c892cb4 files · +153 −53Browse files

Unified · Split

4 files

Source/Libraries/GSF.Core.Shared/Threading/LogicalThread.cs +65 −26
Source/Libraries/GSF.Core.Shared/Threading/LogicalThreadScheduler.cs +83 −22
Source/Libraries/GSF.Core/IO/FileProcessor.cs +2 −2
Source/Libraries/GSF.ServiceProcess/ServiceHelper.cs +3 −3

Source/Libraries/GSF.Core.Shared/Threading/LogicalThread.cs+65−26

         private LogicalThreadScheduler m_scheduler;
         private ConcurrentQueue<Action>[] m_queues;
         private Dictionary<object, object> m_threadLocalStorage;
-        private int m_isActive;
+        private ICancellationToken m_nextExecutionToken;
+        private int m_activePriority;
         private LogicalThreadStatistics m_statistics;
         /// Creates a new instance of the <see cref="LogicalThread"/> class.
         /// </summary>
         public LogicalThread()
-            : this(1)
+            : this(DefaultScheduler)
-        {
-        }
-        /// <summary>
-        /// Creates a new instance of the <see cref="LogicalThread"/> class.
-        /// </summary>
-        /// <param name="priorityLevels">The number of levels of priority supported by this logical thread.</param>
-        public LogicalThread(int priorityLevels)
-            : this(DefaultScheduler, priorityLevels)
         {
         }
         /// Creates a new instance of the <see cref="LogicalThread"/> class.
         /// </summary>
         /// <param name="scheduler">The <see cref="LogicalThreadScheduler"/> that created this thread.</param>
-        /// <param name="priorityLevels">The number of levels of priority supported by this logical thread.</param>
         /// <exception cref="ArgumentNullException"><paramref name="scheduler"/> is null.</exception>
-        /// <exception cref="ArgumentException"><paramref name="priorityLevels"/> is less than or equal to zero.</exception>
+        internal LogicalThread(LogicalThreadScheduler scheduler)
-        internal LogicalThread(LogicalThreadScheduler scheduler, int priorityLevels)
         {
             if ((object)scheduler == null)
                 throw new ArgumentNullException(nameof(scheduler));
-            if (priorityLevels <= 0)
-                throw new ArgumentException($"A logical thread must have at least one priority level.", nameof(priorityLevels));
             m_scheduler = scheduler;
-            m_queues = new ConcurrentQueue<Action>[priorityLevels];
+            m_queues = new ConcurrentQueue<Action>[PriorityLevels];
             m_threadLocalStorage = new Dictionary<object, object>();
+            m_nextExecutionToken = new CancellationToken();
             m_statistics = new LogicalThreadStatistics();
             for (int i = 0; i < m_queues.Length; i++)
         {
             get
             {
-                return m_queues.Length;
+                return m_scheduler.PriorityLevels;
             }
         }
             }
         }
+        /// <summary>
+        /// Gets or sets the priority at which the
+        /// logical thread is queued by the scheduler.
+        /// </summary>
+        internal int ActivePriority
+        {
+            get
+            {
+                return Interlocked.CompareExchange(ref m_activePriority, 0, 0);
+            }
+            set
+            {
+                Interlocked.Exchange(ref m_activePriority, value);
+            }
+        }
+        /// <summary>
+        /// Gets the priority of the next action to be processed on this logical thread.
+        /// </summary>
+        internal int NextPriority
+        {
+            get
+            {
+                return PriorityLevels - m_queues.TakeWhile(queue => queue.IsEmpty).Count();
+            }
+        }
+        /// <summary>
+        /// Gets or sets the cancellation token for the next time
+        /// the thread's actions will be executed by the scheduler.
+        /// </summary>
+        internal ICancellationToken NextExecutionToken
+        {
+            get
+            {
+                return Interlocked.CompareExchange(ref m_nextExecutionToken, null, null);
+            }
+            set
+            {
+                Interlocked.Exchange(ref m_nextExecutionToken, value);
+            }
+        }
         #endregion
         #region [ Methods ]
                 throw new ArgumentException($"Priority {priority} is outside the range between 1 and {PriorityLevels}.", nameof(priority));
             m_queues[PriorityLevels - priority].Enqueue(action);
-            m_scheduler.SignalItemHandler(this);
+            m_scheduler.SignalItemHandler(this, priority);
         }
         /// <summary>
         }
         /// <summary>
-        /// Activates the thread if it is not already active.
+        /// Attempts to activate the thread at the given priority.
         /// </summary>
-        /// <returns>True if this call to TryActivate caused the thread to be active.</returns>
+        /// <param name="priority">The priority at which to activate the thread.</param>
-        internal bool TryActivate()
+        /// <returns>True if the thread's priority needs to be changed to the given priority; false otherwise.</returns>
+        internal bool TryActivate(int priority)
         {
-            return Interlocked.CompareExchange(ref m_isActive, 1, 0) == 0;
+            // Always get the execution token before the
+            // active priority to mitigate race conditions
+            ICancellationToken nextExecutionToken = NextExecutionToken;
+            int activePriority = ActivePriority;
+            return (activePriority < priority) && nextExecutionToken.Cancel();
         }
         /// <summary>
-        /// Deactivates the thread.
+        /// Completely deactivates the thread, making it available for reactivation at any priority.
         /// </summary>
         internal void Deactivate()
         {
-            Interlocked.Exchange(ref m_isActive, 0);
+            // Always update the active priority before the
+            // execution token to mitigate race conditions
+            ActivePriority = 0;
+            NextExecutionToken = new CancellationToken();
         }
         /// <summary>

Source/Libraries/GSF.Core.Shared/Threading/LogicalThreadScheduler.cs+83−22

         public event EventHandler<EventArgs<Exception>> UnhandledException;
         // Fields
-        private ConcurrentQueue<LogicalThread> m_logicalThreads;
+        private ConcurrentQueue<Func<LogicalThread>>[] m_logicalThreadQueues;
         private int m_maxThreadCount;
         private int m_threadCount;
         private bool m_useBackgroundThreads;
         /// <summary>
         /// Creates a new instance of the <see cref="LogicalThreadScheduler"/> class.
         /// </summary>
-        public LogicalThreadScheduler()
+        /// <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_logicalThreads = new ConcurrentQueue<LogicalThread>();
+            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
                     throw new ArgumentException("Max thread count must be greater than zero", nameof(value));
                 int diff = value - m_maxThreadCount;
-                int inactiveThreads = Math.Min(diff, m_logicalThreads.Count);
+                int inactiveThreads = Math.Min(diff, m_logicalThreadQueues.Sum(queue => queue.Count));
                 Interlocked.Exchange(ref m_maxThreadCount, 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>
         /// <returns>A new logical thread managed by this scheduler.</returns>
         public LogicalThread CreateThread()
         {
-            return CreateThread(1);
+            return new LogicalThread(this);
-        }
-        /// <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>
         /// 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)
+        /// <param name="priority">The priority at which the thread is being signaled.</param>
+        internal void SignalItemHandler(LogicalThread thread, int priority)
         {
-            if (thread.TryActivate())
+            if (thread.TryActivate(priority))
             {
-                m_logicalThreads.Enqueue(thread);
+                Enqueue(thread);
                 ActivatePhysicalThread();
             }
         }
         private void ProcessLogicalThreads()
         {
             Stopwatch stopwatch;
+            Func<LogicalThread> accessor = null;
             LogicalThread thread;
             Action action;
             stopwatch = new Stopwatch();
-            while (ThreadCount <= MaxThreadCount && m_logicalThreads.TryDequeue(out thread))
+            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)
                 }
                 if (thread.HasAction)
-                    m_logicalThreads.Enqueue(thread);
+                    Enqueue(thread);
                 else
                     thread.Deactivate();
             }
             DeactivatePhysicalThread();
-            if (!m_logicalThreads.IsEmpty)
+            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>

Source/Libraries/GSF.Core/IO/FileProcessor.cs+2−2

@@ -776,9 +776,9 @@ namespace GSF.IO · 775 unmodified lines
776		776
777	m_fileWatchersLock = new object();	777	m_fileWatchersLock = new object();
778	m_fileWatchers = new List<SafeFileWatcher>();	778	m_fileWatchers = new List<SafeFileWatcher>();
779	m_threadScheduler = new LogicalThreadScheduler();	779	m_threadScheduler = new LogicalThreadScheduler(2);
780	m_threadScheduler.UnhandledException += (sender, args) => OnError(args.Argument);	780	m_threadScheduler.UnhandledException += (sender, args) => OnError(args.Argument);
781	m_processingThread = m_threadScheduler.CreateThread(2);	781	m_processingThread = m_threadScheduler.CreateThread();
782	m_watcherThread = m_threadScheduler.CreateThread();	782	m_watcherThread = m_threadScheduler.CreateThread();
783	m_fileWatchTimer = new Timer(15000);	783	m_fileWatchTimer = new Timer(15000);
784	m_fileWatchTimer.Elapsed += FileWatchTimer_Elapsed;	784	m_fileWatchTimer.Elapsed += FileWatchTimer_Elapsed;

Source/Libraries/GSF.ServiceProcess/ServiceHelper.cs+3−3

@@ -235,7 +235,7 @@ namespace GSF.ServiceProcess · 234 unmodified lines
235	{	235	{
236	m_clientID = clientID;	236	m_clientID = clientID;
237	m_filter = new ClientFilter();	237	m_filter = new ClientFilter();
238	m_thread = serviceHelper.m_threadScheduler.CreateThread(2);	238	m_thread = serviceHelper.m_threadScheduler.CreateThread();
239	m_messageCounts = new List<Tuple<DateTime, int>>();	239	m_messageCounts = new List<Tuple<DateTime, int>>();
240	m_serviceHelper = serviceHelper;	240	m_serviceHelper = serviceHelper;
241	}	241	}
@@ -720,9 +720,9 @@ namespace GSF.ServiceProcess · 478 unmodified lines
720	m_componentEnabledStates = new Dictionary<ISupportLifecycle, bool>();	720	m_componentEnabledStates = new Dictionary<ISupportLifecycle, bool>();
721		721
722	m_clientStatusUpdateLookup = new Dictionary<Guid, ClientStatusUpdateConfiguration>();	722	m_clientStatusUpdateLookup = new Dictionary<Guid, ClientStatusUpdateConfiguration>();
723	m_threadScheduler = new LogicalThreadScheduler();	723	m_threadScheduler = new LogicalThreadScheduler(2);
724	m_threadScheduler.UnhandledException += LogicalThread_ProcessException;	724	m_threadScheduler.UnhandledException += LogicalThread_ProcessException;
725	m_statusUpdateThread = m_threadScheduler.CreateThread(2);	725	m_statusUpdateThread = m_threadScheduler.CreateThread();
726	m_statusUpdateQueue = new List<StatusUpdate>();	726	m_statusUpdateQueue = new List<StatusUpdate>();
727		727
728	// Components	728	// Components

GSF.Core: Updated LogicalThread API to handle prioritization at the thread scheduler so that threads can be reprioritized based on the priority of their tasks.

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code