//******************************************************************************************************// LogicalThread.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/24/2015 - Stephen C. Wills// Generated original version of source code.// 02/16/2016 - J. Ritchie Carroll// Changed default priority level to 1 - normal / lowest.////****************************************************************************************************** using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Linq;using System.Threading; namespace GSF.Threading{ /// <summary> /// Represents a set of statistics gathered about /// the execution time of actions on a logical thread. /// </summary> public class LogicalThreadStatistics { #region [ Members ] // Fields private TimeSpan m_maxExecutionTime; private TimeSpan m_minExecutionTime; private TimeSpan m_totalExecutionTime; private long m_executionCount; #endregion #region [ Properties ] /// <summary> /// Gets the execution time of the longest running action. /// </summary> public TimeSpan MaxExecutionTime { get { return m_maxExecutionTime; } } /// <summary> /// Gets the execution time of the shortest running action. /// </summary> public TimeSpan MinExecutionTime { get { return m_minExecutionTime; } } /// <summary> /// Gets the total execution time of all actions executed on the logical thread. /// </summary> public TimeSpan TotalExecutionTime { get { return m_totalExecutionTime; } } /// <summary> /// Gets the total number of actions executed on the logical thread. /// </summary> public long ExecutionCount { get { return m_executionCount; } } #endregion #region [ Methods ] /// <summary> /// Updates the statistics based on the execution time of a single action. /// </summary> /// <param name="executionTime">The execution time of the action.</param> internal void UpdateStatistics(TimeSpan executionTime) { if (m_executionCount == 0 || executionTime > m_maxExecutionTime) m_maxExecutionTime = executionTime; if (m_executionCount == 0 || executionTime < m_minExecutionTime) m_minExecutionTime = executionTime; m_totalExecutionTime += executionTime; m_executionCount++; } #endregion } /// <summary> /// Represents a thread of execution to which /// actions can be dispatched from other threads. /// </summary> /// <remarks> /// <para> /// This class provides a simple alternative to synchronization primitives /// such as wait handles and locks. Actions dispatched to a logical thread /// will be processed synchronously as though it was executed as consecutive /// method calls. All such actions can be dispatched from any thread in the /// system so method calls coming from multiple threads can be easily /// synchronized without locks, loops, wait handles, or timeouts. /// </para> /// /// <para> /// Note that the <see cref="LogicalThreadScheduler"/> implements its own /// thread pool to execute tasks pushed to logical threads. Executing /// long-running processes or using synchronization primitives with high /// contention or long timeouts can hinder the logical thread scheduler's /// ability to schedule the actions of other logical threads. Like other /// thread pool implementations, you can mitigate this by increasing the /// maximum thread count of the logical thread scheduler, however it is /// recommended to avoid using synchronization primitives and instead /// synchronize those operations by running them as separate actions on /// the same logical thread. /// </para> /// </remarks> public class LogicalThread { #region [ Members ] // Events /// <summary> /// Handler for unhandled exceptions on the thread. /// </summary> public event EventHandler<EventArgs<Exception>> UnhandledException; // Fields private LogicalThreadScheduler m_scheduler; private ConcurrentQueue<Action>[] m_queues; private Dictionary<object, object> m_threadLocalStorage; private int m_isActive; private LogicalThreadStatistics m_statistics; #endregion #region [ Constructors ] /// <summary> /// Creates a new instance of the <see cref="LogicalThread"/> class. /// </summary> public LogicalThread() : this(1) { } /// <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) { } /// <summary> /// 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, 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_threadLocalStorage = new Dictionary<object, object>(); m_statistics = new LogicalThreadStatistics(); for (int i = 0; i < m_queues.Length; i++) m_queues[i] = new ConcurrentQueue<Action>(); } #endregion #region [ Properties ] /// <summary> /// Gets the number of levels of priority /// supported by this logical thread. /// </summary> public int PriorityLevels { get { return m_queues.Length; } } /// <summary> /// Gets a flag that indicates whether the logical /// thread has any unprocessed actions left in its queue. /// </summary> public bool HasAction { get { return m_queues.Any(queue => !queue.IsEmpty); } } #endregion #region [ Methods ] /// <summary> /// Pushes an action to the logical thread. /// </summary> /// <param name="action">The action to be executed on this thread.</param> public void Push(Action action) { Push(1, action); } /// <summary> /// Pushes an action to the logical thread. /// </summary> /// <param name="priority">The priority the action should be given when executing actions on the thread (higher numbers have higher priority).</param> /// <param name="action">The action to be executed on this thread.</param> /// <exception cref="ArgumentException"><paramref name="priority"/> is outside the range between 1 and <see cref="PriorityLevels"/>.</exception> public void Push(int priority, Action action) { if (priority < 1 || priority > PriorityLevels) 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); } /// <summary> /// Clears all actions from the logical thread. /// </summary> public void Clear() { Action action; foreach (ConcurrentQueue<Action> queue in m_queues) { while (!queue.IsEmpty) queue.TryDequeue(out action); } } /// <summary> /// Samples the statistics, providing current statistic /// values and resetting statistic counters. /// </summary> /// <returns>The current statistic values.</returns> public LogicalThreadStatistics SampleStatistics() { return Interlocked.Exchange(ref m_statistics, new LogicalThreadStatistics()); } /// <summary> /// Pulls an action from the logical thread's internal queue to be executed on a physical thread. /// </summary> /// <returns>An action from the logical thread's internal queue.</returns> internal Action Pull() { Action action; foreach (ConcurrentQueue<Action> queue in m_queues) { if (queue.TryDequeue(out action)) return action; } return null; } /// <summary> /// Activates the thread if it is not already active. /// </summary> /// <returns>True if this call to TryActivate caused the thread to be active.</returns> internal bool TryActivate() { return Interlocked.CompareExchange(ref m_isActive, 1, 0) == 0; } /// <summary> /// Deactivates the thread. /// </summary> internal void Deactivate() { Interlocked.Exchange(ref m_isActive, 0); } /// <summary> /// Returns the thread local object with the given ID. /// </summary> /// <param name="key">The key used to look up the thread local object.</param> /// <returns>The value of the thread local object with the given ID.</returns> internal object GetThreadLocal(object key) { object value; m_threadLocalStorage.TryGetValue(key, out value); return value; } /// <summary> /// Sets the value of the thread local object with the given ID. /// </summary> /// <param name="key">The key used to look up the thread local object.</param> /// <param name="value">The new value for the thread local object.</param> internal void SetThreadLocal(object key, object value) { if (value != null) m_threadLocalStorage[key] = value; else m_threadLocalStorage.Remove(key); } /// <summary> /// Updates the statistics based on the execution time of a single action. /// </summary> /// <param name="executionTime">The execution time of the action.</param> internal void UpdateStatistics(TimeSpan executionTime) { m_statistics.UpdateStatistics(executionTime); } /// <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> internal bool OnUnhandledException(Exception ex) { EventHandler<EventArgs<Exception>> unhandledException = UnhandledException; if ((object)unhandledException == null) return false; unhandledException(this, new EventArgs<Exception>(ex)); return true; } #endregion #region [ Static ] // Static Fields private static readonly LogicalThreadScheduler DefaultScheduler = new LogicalThreadScheduler(); private static readonly ThreadLocal<LogicalThread> LocalThread = new ThreadLocal<LogicalThread>(); // Static Properties /// <summary> /// Gets the logical thread that is currently executing. /// </summary> public static LogicalThread CurrentThread { get { return LocalThread.Value; } internal set { LocalThread.Value = value; } } #endregion }}
//******************************************************************************************************
// LogicalThread.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/24/2015 - Stephen C. Wills
// Generated original version of source code.
// 02/16/2016 - J. Ritchie Carroll
// Changed default priority level to 1 - normal / lowest.
//
//******************************************************************************************************
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
namespace GSF.Threading
{
/// <summary>
/// Represents a set of statistics gathered about
/// the execution time of actions on a logical thread.
/// </summary>
public class LogicalThreadStatistics
{
#region [ Members ]
// Fields
private TimeSpan m_maxExecutionTime;
private TimeSpan m_minExecutionTime;
private TimeSpan m_totalExecutionTime;
private long m_executionCount;
#endregion
#region [ Properties ]
/// <summary>
/// Gets the execution time of the longest running action.
/// </summary>
public TimeSpan MaxExecutionTime
{
get
{
return m_maxExecutionTime;
}
}
/// <summary>
/// Gets the execution time of the shortest running action.
/// </summary>
public TimeSpan MinExecutionTime
{
get
{
return m_minExecutionTime;
}
}
/// <summary>
/// Gets the total execution time of all actions executed on the logical thread.
/// </summary>
public TimeSpan TotalExecutionTime
{
get
{
return m_totalExecutionTime;
}
}
/// <summary>
/// Gets the total number of actions executed on the logical thread.
/// </summary>
public long ExecutionCount
{
get
{
return m_executionCount;
}
}
#endregion
#region [ Methods ]
/// <summary>
/// Updates the statistics based on the execution time of a single action.
/// </summary>
/// <param name="executionTime">The execution time of the action.</param>
internal void UpdateStatistics(TimeSpan executionTime)
{
if (m_executionCount == 0 || executionTime > m_maxExecutionTime)
m_maxExecutionTime = executionTime;
if (m_executionCount == 0 || executionTime < m_minExecutionTime)
m_minExecutionTime = executionTime;
m_totalExecutionTime += executionTime;
m_executionCount++;
}
#endregion
}
/// <summary>
/// Represents a thread of execution to which
/// actions can be dispatched from other threads.
/// </summary>
/// <remarks>
/// <para>
/// This class provides a simple alternative to synchronization primitives
/// such as wait handles and locks. Actions dispatched to a logical thread
/// will be processed synchronously as though it was executed as consecutive
/// method calls. All such actions can be dispatched from any thread in the
/// system so method calls coming from multiple threads can be easily
/// synchronized without locks, loops, wait handles, or timeouts.
/// </para>
///
/// <para>
/// Note that the <see cref="LogicalThreadScheduler"/> implements its own
/// thread pool to execute tasks pushed to logical threads. Executing
/// long-running processes or using synchronization primitives with high
/// contention or long timeouts can hinder the logical thread scheduler's
/// ability to schedule the actions of other logical threads. Like other
/// thread pool implementations, you can mitigate this by increasing the
/// maximum thread count of the logical thread scheduler, however it is
/// recommended to avoid using synchronization primitives and instead
/// synchronize those operations by running them as separate actions on
/// the same logical thread.
/// </para>
/// </remarks>
public class LogicalThread
{
#region [ Members ]
// Events
/// <summary>
/// Handler for unhandled exceptions on the thread.
/// </summary>
public event EventHandler<EventArgs<Exception>> UnhandledException;
// Fields
private LogicalThreadScheduler m_scheduler;
private ConcurrentQueue<Action>[] m_queues;
private Dictionary<object, object> m_threadLocalStorage;
private int m_isActive;
private LogicalThreadStatistics m_statistics;
#endregion
#region [ Constructors ]
/// <summary>
/// Creates a new instance of the <see cref="LogicalThread"/> class.
/// </summary>
public LogicalThread()
: this(1)
{
}
/// <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)
{
}
/// <summary>
/// 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, 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_threadLocalStorage = new Dictionary<object, object>();
m_statistics = new LogicalThreadStatistics();
for (int i = 0; i < m_queues.Length; i++)
m_queues[i] = new ConcurrentQueue<Action>();
}
#endregion
#region [ Properties ]
/// <summary>
/// Gets the number of levels of priority
/// supported by this logical thread.
/// </summary>
public int PriorityLevels
{
get
{
return m_queues.Length;
}
}
/// <summary>
/// Gets a flag that indicates whether the logical
/// thread has any unprocessed actions left in its queue.
/// </summary>
public bool HasAction
{
get
{
return m_queues.Any(queue => !queue.IsEmpty);
}
}
#endregion
#region [ Methods ]
/// <summary>
/// Pushes an action to the logical thread.
/// </summary>
/// <param name="action">The action to be executed on this thread.</param>
public void Push(Action action)
{
Push(1, action);
}
/// <summary>
/// Pushes an action to the logical thread.
/// </summary>
/// <param name="priority">The priority the action should be given when executing actions on the thread (higher numbers have higher priority).</param>
/// <param name="action">The action to be executed on this thread.</param>
/// <exception cref="ArgumentException"><paramref name="priority"/> is outside the range between 1 and <see cref="PriorityLevels"/>.</exception>
public void Push(int priority, Action action)
{
if (priority < 1 || priority > PriorityLevels)
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);
}
/// <summary>
/// Clears all actions from the logical thread.
/// </summary>
public void Clear()
{
Action action;
foreach (ConcurrentQueue<Action> queue in m_queues)
{
while (!queue.IsEmpty)
queue.TryDequeue(out action);
}
}
/// <summary>
/// Samples the statistics, providing current statistic
/// values and resetting statistic counters.
/// </summary>
/// <returns>The current statistic values.</returns>
public LogicalThreadStatistics SampleStatistics()
{
return Interlocked.Exchange(ref m_statistics, new LogicalThreadStatistics());
}
/// <summary>
/// Pulls an action from the logical thread's internal queue to be executed on a physical thread.
/// </summary>
/// <returns>An action from the logical thread's internal queue.</returns>
internal Action Pull()
{
Action action;
foreach (ConcurrentQueue<Action> queue in m_queues)
{
if (queue.TryDequeue(out action))
return action;
}
return null;
}
/// <summary>
/// Activates the thread if it is not already active.
/// </summary>
/// <returns>True if this call to TryActivate caused the thread to be active.</returns>
internal bool TryActivate()
{
return Interlocked.CompareExchange(ref m_isActive, 1, 0) == 0;
}
/// <summary>
/// Deactivates the thread.
/// </summary>
internal void Deactivate()
{
Interlocked.Exchange(ref m_isActive, 0);
}
/// <summary>
/// Returns the thread local object with the given ID.
/// </summary>
/// <param name="key">The key used to look up the thread local object.</param>
/// <returns>The value of the thread local object with the given ID.</returns>
internal object GetThreadLocal(object key)
{
object value;
m_threadLocalStorage.TryGetValue(key, out value);
return value;
}
/// <summary>
/// Sets the value of the thread local object with the given ID.
/// </summary>
/// <param name="key">The key used to look up the thread local object.</param>
/// <param name="value">The new value for the thread local object.</param>
internal void SetThreadLocal(object key, object value)
{
if (value != null)
m_threadLocalStorage[key] = value;
else
m_threadLocalStorage.Remove(key);
}
/// <summary>
/// Updates the statistics based on the execution time of a single action.
/// </summary>
/// <param name="executionTime">The execution time of the action.</param>
internal void UpdateStatistics(TimeSpan executionTime)
{
m_statistics.UpdateStatistics(executionTime);
}
/// <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>
internal bool OnUnhandledException(Exception ex)
{
EventHandler<EventArgs<Exception>> unhandledException = UnhandledException;
if ((object)unhandledException == null)
return false;
unhandledException(this, new EventArgs<Exception>(ex));
return true;
}
#endregion
#region [ Static ]
// Static Fields
private static readonly LogicalThreadScheduler DefaultScheduler = new LogicalThreadScheduler();
private static readonly ThreadLocal<LogicalThread> LocalThread = new ThreadLocal<LogicalThread>();
// Static Properties
/// <summary>
/// Gets the logical thread that is currently executing.
/// </summary>
public static LogicalThread CurrentThread
{
get
{
return LocalThread.Value;
}
internal set
{
LocalThread.Value = value;
}
}
#endregion
}
}