dotnet
/
corert
зеркало из
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

Enable CLR Thread Pool via a MSBuild flag (default enabled on Unix) 

This PR completes, adds tests for, and enables the CLR Thread Pool on Unix. It also adds Windows implementations for the low level primitives used in the CLR Thread Pool.

The thread pool is enabled by setting the MSBuild property FeaturePortableThreadPool to true, which is the default on Unix.

Supersedes #4039.
Rebased and squashed replacement for original PR #4124.
This commit is contained in:
t-jekor 2017-07-05 10:44:15 -07:00 коммит произвёл Jan Kotas
Родитель a410be2def
Коммит 18eab60696
22 изменённых файлов: 838 добавлений и 193 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

23
src/Common/src/Interop/Windows/kernel32/Interop.CompletionPort.cs Normal file
Просмотреть файл

@ -0,0 +1,23 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Runtime.InteropServices;
internal partial class Interop
{
internal partial class Kernel32
{
[DllImport(Libraries.Kernel32, SetLastError = true)]
internal static extern IntPtr CreateIoCompletionPort(IntPtr FileHandle, IntPtr ExistingCompletionPort, UIntPtr CompletionKey, int NumberOfConcurrentThreads);
[DllImport(Libraries.Kernel32, SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
internal static extern bool PostQueuedCompletionStatus(IntPtr CompletionPort, int dwNumberOfBytesTransferred, UIntPtr CompletionKey, IntPtr lpOverlapped);
[DllImport(Libraries.Kernel32, SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
internal static extern bool GetQueuedCompletionStatus(IntPtr CompletionPort, out int lpNumberOfBytes, out UIntPtr CompletionKey, out IntPtr lpOverlapped, int dwMilliseconds);
}
}

28
src/Common/src/Interop/Windows/kernel32/Interop.ConditionVariable.cs Normal file
Просмотреть файл

@ -0,0 +1,28 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Runtime.InteropServices;
internal partial class Interop
{
internal partial class Kernel32
{
[StructLayout(LayoutKind.Sequential)]
internal struct CONDITION_VARIABLE
{
private IntPtr Ptr;
}
[DllImport(Libraries.Kernel32)]
internal static extern void InitializeConditionVariable(out CONDITION_VARIABLE ConditionVariable);
[DllImport(Libraries.Kernel32, SetLastError = true)]
internal static extern void WakeConditionVariable(ref CONDITION_VARIABLE ConditionVariable);
[DllImport(Libraries.Kernel32, SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
internal static extern bool SleepConditionVariableCS(ref CONDITION_VARIABLE ConditionVariable, ref CRITICAL_SECTION CriticalSection, int dwMilliseconds);
}
}

35
src/Common/src/Interop/Windows/kernel32/Interop.CriticalSection.cs Normal file
Просмотреть файл

@ -0,0 +1,35 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Runtime.InteropServices;
internal partial class Interop
{
internal partial class Kernel32
{
[StructLayout(LayoutKind.Sequential)]
internal struct CRITICAL_SECTION
{
private IntPtr DebugInfo;
private int LockCount;
private int RecursionCount;
private IntPtr OwningThread;
private IntPtr LockSemaphore;
private UIntPtr SpinCount;
}
[DllImport(Libraries.Kernel32)]
internal static extern void InitializeCriticalSection(out CRITICAL_SECTION lpCriticalSection);
[DllImport(Libraries.Kernel32)]
internal static extern void EnterCriticalSection(ref CRITICAL_SECTION lpCriticalSection);
[DllImport(Libraries.Kernel32)]
internal static extern void LeaveCriticalSection(ref CRITICAL_SECTION lpCriticalSection);
[DllImport(Libraries.Kernel32)]
internal static extern void DeleteCriticalSection(ref CRITICAL_SECTION lpCriticalSection);
}
}

16
src/Common/src/Interop/Windows/kernel32/Interop.GetSystemTimes.cs Normal file
Просмотреть файл

@ -0,0 +1,16 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Runtime.InteropServices;
internal partial class Interop
{
internal partial class Kernel32
{
[DllImport(Libraries.Kernel32, SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
internal static extern bool GetSystemTimes(out ulong idleTime, out ulong kernelTime, out ulong userTime);
}
}

17
src/Common/src/Interop/Windows/kernel32/Interop.QueryPerformance.cs Normal file
Просмотреть файл

@ -0,0 +1,17 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Runtime.InteropServices;
internal partial class Interop
{
internal partial class Kernel32
{
[DllImport(Libraries.Kernel32)]
internal static extern bool QueryPerformanceCounter(out ulong value);
[DllImport(Libraries.Kernel32)]
internal static extern bool QueryPerformanceFrequency(out ulong value);
}
}

1
src/System.Private.CoreLib/shared/Interop/Windows/Interop.Errors.cs
Просмотреть файл

@ -40,5 +40,6 @@ internal partial class Interop
internal const int ERROR_NOT_FOUND = 0x490;
internal const int ERROR_BAD_IMPERSONATION_LEVEL = 0x542;
internal const int E_FILENOTFOUND = unchecked((int)0x80070002);
internal const int ERROR_TIMEOUT = 0x000005B4;
}
}

50
src/System.Private.CoreLib/src/System.Private.CoreLib.csproj
Просмотреть файл

@ -38,6 +38,10 @@
<PropertyGroup>
<DefineConstants Condition="'$(FeatureCominterop)' == 'true'">FEATURE_COMINTEROP;$(DefineConstants)</DefineConstants>
</PropertyGroup>
<PropertyGroup>
<FeaturePortableThreadPool Condition="'$(FeaturePortableThreadPool)' == ''">false</FeaturePortableThreadPool>
<FeaturePortableThreadPool Condition="'$(TargetsUnix)' == 'true'">true</FeaturePortableThreadPool>
</PropertyGroup>
<ItemGroup>
<Compile Include="..\..\Common\src\Internal\Runtime\RuntimeConstants.cs">
<Link>Internal\Runtime\RuntimeConstants.cs</Link>
@ -326,6 +330,8 @@
<Compile Include="System\Threading\Interlocked.cs" />
<Compile Include="System\Threading\IOCompletionCallback.cs" />
<Compile Include="System\Threading\LockHolder.cs" />
<Compile Include="System\Threading\LowLevelLock.cs" />
<Compile Include="System\Threading\LowLevelMonitor.cs" />
<Compile Include="System\Threading\ManualResetEventSlim.cs" />
<Compile Include="System\Threading\Monitor.cs" />
<Compile Include="System\Threading\Mutex.cs" />
@ -457,9 +463,24 @@
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.MultiByteToWideChar.cs">
<Link>Interop\Windows\kernel32\Interop.MultiByteToWideChar.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.CompletionPort.cs">
<Link>Interop\Windows\kernel32\Interop.CompletionPort.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.CriticalSection.cs">
<Link>Interop\Windows\kernel32\Interop.CriticalSection.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.ConditionVariable.cs">
<Link>Interop\Windows\kernel32\Interop.ConditionVariable.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.GetSystemTimes.cs">
<Link>Interop\Windows\kernel32\Interop.GetSystemTimes.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.GetModuleFileName.cs">
<Link>Interop\Windows\kernel32\Interop.GetModuleFileName.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\kernel32\Interop.QueryPerformance.cs">
<Link>Interop\Windows\kernel32\Interop.QueryPerformance.cs</Link>
</Compile>
<Compile Include="..\..\Common\src\Interop\Windows\mincore\Interop.GetCurrentThreadId.cs">
<Link>Interop\Windows\mincore\Interop.GetCurrentThreadId.cs</Link>
</Compile>
@ -514,10 +535,13 @@
<Compile Condition="'$(EnableWinRT)' != 'true'" Include="..\..\Common\src\Interop\Windows\mincore\Interop.DynamicLoad.cs">
<Link>Interop\Windows\mincore\Interop.DynamicLoad.cs</Link>
</Compile>
<Compile Include="System\HighPerformanceCounter.Windows.cs" />
<Compile Include="System\Threading\EventWaitHandle.Windows.cs" />
<Compile Include="System\Threading\LowLevelMonitor.Windows.cs" />
<Compile Include="System\Threading\LowLevelLifoSemaphore.Windows.cs" />
<Compile Include="System\Threading\Mutex.Windows.cs" />
<Compile Include="System\Threading\Semaphore.Windows.cs" />
<Compile Include="System\Threading\ThreadPool.Windows.cs" />
<Compile Include="System\Threading\ThreadPool.Windows.cs" Condition="'$(FeaturePortableThreadPool)' != 'true'" />
<Compile Include="System\Threading\WaitHandle.Windows.cs" />
<Compile Include="System\Threading\Win32ThreadPoolBoundHandle.cs" />
<Compile Include="System\Threading\Win32ThreadPoolNativeOverlapped.cs" />
@ -543,6 +567,19 @@
<Link>Interop\Windows\mincore\Interop.Normalization.cs</Link>
</Compile>
</ItemGroup>
<ItemGroup Condition="'$(FeaturePortableThreadPool)' == 'true'">
<Compile Include="System\Threading\ThreadPool.Portable.cs" />
<Compile Include="System\Threading\ClrThreadPool.cs" />
<Compile Include="System\Threading\ClrThreadPoolEventSource.cs" />
<Compile Include="System\Threading\ClrThreadPool.GateThread.cs" />
<Compile Include="System\Threading\ClrThreadPool.HillClimbing.cs" />
<Compile Include="System\Threading\ClrThreadPool.HillClimbing.Complex.cs" />
<Compile Include="System\Threading\ClrThreadPool.ThreadCounts.cs" />
<Compile Include="System\Threading\ClrThreadPool.WaitThread.cs" />
<Compile Include="System\Threading\ClrThreadPool.WorkerThread.cs" />
<Compile Include="System\Threading\ClrThreadPool.CpuUtilizationReader.Unix.cs" Condition="'$(TargetsUnix)' == 'true'" />
<Compile Include="System\Threading\ClrThreadPool.CpuUtilizationReader.Windows.cs" Condition="'$(TargetsWindows)'=='true'" />
</ItemGroup>
<ItemGroup>
<!-- CORERT-TODO: Port to Unix -->
<Compile Include="..\..\Common\src\Interop\Windows\Interop.Libraries.cs">
@ -600,22 +637,11 @@
<Compile Include="System\Number.Unix.cs" />
<Compile Include="System\HighPerformanceCounter.Unix.cs" />
<Compile Include="System\Runtime\InteropServices\PInvokeMarshal.Unix.cs" />
<Compile Include="System\Threading\ClrThreadPool.cs" />
<Compile Include="System\Threading\ClrThreadPool.CpuUtilizationReader.Unix.cs" />
<Compile Include="System\Threading\ClrThreadPool.GateThread.cs" />
<Compile Include="System\Threading\ClrThreadPool.HillClimbing.cs" />
<Compile Include="System\Threading\ClrThreadPool.HillClimbing.Complex.cs" />
<Compile Include="System\Threading\ClrThreadPool.ThreadCounts.cs" />
<Compile Include="System\Threading\ClrThreadPool.WaitThread.cs" />
<Compile Include="System\Threading\ClrThreadPool.WorkerThread.cs" />
<Compile Include="System\Threading\ClrThreadPoolEventSource.cs" />
<Compile Include="System\Threading\EventWaitHandle.Unix.cs" />
<Compile Include="System\Threading\LowLevelLifoSemaphore.Unix.cs" />
<Compile Include="System\Threading\LowLevelLock.Unix.cs" />
<Compile Include="System\Threading\LowLevelMonitor.Unix.cs" />
<Compile Include="System\Threading\Mutex.Unix.cs" />
<Compile Include="System\Threading\Semaphore.Unix.cs" />
<Compile Include="System\Threading\ThreadPool.Unix.cs" />
<Compile Include="System\Threading\Timer.Unix.cs" />
<Compile Include="System\Threading\WaitHandle.Unix.cs" />
<Compile Include="System\Threading\WaitSubsystem.HandleManager.Unix.cs" />

3
src/System.Private.CoreLib/src/System/HighPerformanceCounter.Unix.cs
Просмотреть файл

@ -9,7 +9,6 @@ namespace System
public static ulong TickCount => Interop.Sys.GetHighPrecisionCount();
// Cache the frequency on the managed side to avoid the cost of P/Invoke on every access to Frequency
private static ulong s_frequency;
public static ulong Frequency => s_frequency == 0 ? s_frequency = Interop.Sys.GetHighPrecisionCounterFrequency() : s_frequency;
public static ulong Frequency { get; } = Interop.Sys.GetHighPrecisionCounterFrequency();
}
}

26
src/System.Private.CoreLib/src/System/HighPerformanceCounter.Windows.cs Normal file
Просмотреть файл

@ -0,0 +1,26 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace System
{
internal static class HighPerformanceCounter
{
public static ulong TickCount
{
get
{
Interop.Kernel32.QueryPerformanceCounter(out ulong counter);
return counter;
}
}
public static ulong Frequency { get; } = GetFrequency();
private static ulong GetFrequency()
{
Interop.Kernel32.QueryPerformanceFrequency(out ulong frequency);
return frequency;
}
}
}

5
src/System.Private.CoreLib/src/System/Threading/ClrThreadPool.CpuUtilizationReader.Unix.cs
Просмотреть файл

@ -9,11 +9,6 @@ namespace System.Threading
private class CpuUtilizationReader
{
private Interop.Sys.ProcessCpuInformation _cpuInfo;
public CpuUtilizationReader()
{
_cpuInfo = new Interop.Sys.ProcessCpuInformation();
Interop.Sys.GetCpuUtilization(ref _cpuInfo); // Initialize the cpuInfo structure so future calls with it get correct readings
}
public int CurrentUtilization => Interop.Sys.GetCpuUtilization(ref _cpuInfo); // Updates cpuInfo as a side effect for the next call
}

54
src/System.Private.CoreLib/src/System/Threading/ClrThreadPool.CpuUtilizationReader.Windows.cs Normal file
Просмотреть файл

@ -0,0 +1,54 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics;
using System.Runtime.InteropServices;
namespace System.Threading
{
internal partial class ClrThreadPool
{
private class CpuUtilizationReader
{
private struct ProcessCpuInformation
{
public long idleTime;
public long kernelTime;
public long userTime;
}
private ProcessCpuInformation _processCpuInfo = new ProcessCpuInformation();
public int CurrentUtilization
{
get
{
if (!Interop.Kernel32.GetSystemTimes(out var idleTime, out var kernelTime, out var userTime))
{
int error = Marshal.GetLastWin32Error();
var exception = new OutOfMemoryException();
exception.SetErrorCode(error);
throw exception;
}
long cpuTotalTime = ((long)userTime - _processCpuInfo.userTime) + ((long)kernelTime - _processCpuInfo.kernelTime);
long cpuBusyTime = cpuTotalTime - ((long)idleTime - _processCpuInfo.idleTime);
_processCpuInfo.kernelTime = (long)kernelTime;
_processCpuInfo.userTime = (long)userTime;
_processCpuInfo.idleTime = (long)idleTime;
if (cpuTotalTime > 0 && cpuBusyTime > 0)
{
long reading = cpuBusyTime * 100 / cpuTotalTime;
reading = Math.Min(reading, 100);
Debug.Assert(0 <= reading);
return (int)reading;
}
return 0;
}
}
}
}
}

8
src/System.Private.CoreLib/src/System/Threading/ClrThreadPool.ThreadCounts.cs
Просмотреть файл

@ -75,10 +75,10 @@ namespace System.Threading
private void Validate()
{
Debug.Assert(numThreadsGoal > 0);
Debug.Assert(numExistingThreads >= 0);
Debug.Assert(numProcessingWork >= 0);
Debug.Assert(numProcessingWork <= numExistingThreads);
Debug.Assert(numThreadsGoal > 0, "Goal must be positive");
Debug.Assert(numExistingThreads >= 0, "Number of existing threads must be non-zero");
Debug.Assert(numProcessingWork >= 0, "Number of threads processing work must be non-zero");
Debug.Assert(numProcessingWork <= numExistingThreads, $"Num processing work ({numProcessingWork}) must be less than or equal to Num existing threads ({numExistingThreads})");
}
}
}

9
src/System.Private.CoreLib/src/System/Threading/ClrThreadPool.WorkerThread.cs
Просмотреть файл

@ -36,15 +36,6 @@ namespace System.Threading
// If the queue runs out of work for us, we need to update the number of working workers to reflect that we are done working for now
RemoveWorkingWorker();
}
// Reset thread-local state that we control.
if (currentThread.Priority != ThreadPriority.Normal)
{
currentThread.Priority = ThreadPriority.Normal;
}
CultureInfo.CurrentCulture = CultureInfo.InstalledUICulture;
CultureInfo.CurrentUICulture = CultureInfo.InstalledUICulture;
}
else
{

6
src/System.Private.CoreLib/src/System/Threading/ClrThreadPool.cs
Просмотреть файл

@ -21,7 +21,7 @@ namespace System.Threading
private const int CpuUtilizationHigh = 95;
private const int CpuUtilizationLow = 80;
private int _cpuUtilization = 85; // TODO: Add calculation for CPU utilization
private int _cpuUtilization = 0;
private static readonly short s_forcedMinWorkerThreads = AppContextConfigHelper.GetInt16Config("System.Threading.ThreadPool.MinThreads", 0);
@ -88,7 +88,7 @@ namespace System.Threading
_minThreads = threads;
ThreadCounts counts = ThreadCounts.VolatileReadCounts(ref _separated.counts);
while (counts.numThreadsGoal < minThreads)
while (counts.numThreadsGoal < _minThreads)
{
ThreadCounts newCounts = counts;
newCounts.numThreadsGoal = _minThreads;
@ -137,7 +137,7 @@ namespace System.Threading
_maxThreads = threads;
ThreadCounts counts = ThreadCounts.VolatileReadCounts(ref _separated.counts);
while (counts.numThreadsGoal > maxThreads)
while (counts.numThreadsGoal > _maxThreads)
{
ThreadCounts newCounts = counts;
newCounts.numThreadsGoal = _maxThreads;

6
src/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.Unix.cs
Просмотреть файл

@ -8,7 +8,7 @@ namespace System.Threading
/// A LIFO semaphore.
/// Waits on this semaphore are uninterruptible.
/// </summary>
internal sealed class LowLevelLifoSemaphore
internal sealed class LowLevelLifoSemaphore : IDisposable
{
private WaitSubsystem.WaitableObject _semaphore;
@ -26,5 +26,9 @@ namespace System.Threading
{
return WaitSubsystem.ReleaseSemaphore(_semaphore, count);
}
public void Dispose()
{
}
}
}

63
src/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.Windows.cs Normal file
Просмотреть файл

@ -0,0 +1,63 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics;
using System.Runtime.InteropServices;
namespace System.Threading
{
/// <summary>
/// A LIFO semaphore implemented using Win32 IO Completion Ports.
/// </summary>
/// <remarks>
/// IO Completion ports release waiting threads in LIFO order, so we can use them to create a LIFO semaphore.
/// See https://msdn.microsoft.com/en-us/library/windows/desktop/aa365198(v=vs.85).aspx under How I/O Completion Ports Work.
/// From the docs "Threads that block their execution on an I/O completion port are released in last-in-first-out (LIFO) order."
/// </remarks>
internal sealed class LowLevelLifoSemaphore : IDisposable
{
private IntPtr _completionPort;
public LowLevelLifoSemaphore(int initialSignalCount, int maximumSignalCount)
{
Debug.Assert(initialSignalCount >= 0, "Windows LowLevelLifoSemaphore does not support a negative signal count"); // TODO: Track actual signal count to enable this
_completionPort = Interop.Kernel32.CreateIoCompletionPort(new IntPtr(-1), IntPtr.Zero, UIntPtr.Zero, 1);
if (_completionPort == IntPtr.Zero)
{
var error = Marshal.GetLastWin32Error();
var exception = new OutOfMemoryException();
exception.SetErrorCode(error);
throw exception;
}
Release(initialSignalCount);
}
public bool Wait(int timeoutMs)
{
bool success = Interop.Kernel32.GetQueuedCompletionStatus(_completionPort, out var numberOfBytes, out var completionKey, out var pointerToOverlapped, timeoutMs);
Debug.Assert(success || (Marshal.GetLastWin32Error() == WaitHandle.WaitTimeout));
return success;
}
public int Release(int count)
{
for (int i = 0; i < count; i++)
{
if(!Interop.Kernel32.PostQueuedCompletionStatus(_completionPort, 1, UIntPtr.Zero, IntPtr.Zero))
{
var lastError = Marshal.GetLastWin32Error();
var exception = new OutOfMemoryException();
exception.SetErrorCode(lastError);
throw exception;
}
}
return 0; // TODO: Track actual signal count to calculate this
}
public void Dispose()
{
Interop.Kernel32.CloseHandle(_completionPort);
}
}
}

0
src/System.Private.CoreLib/src/System/Threading/LowLevelLock.Unix.cs → src/System.Private.CoreLib/src/System/Threading/LowLevelLock.cs
Просмотреть файл

90
src/System.Private.CoreLib/src/System/Threading/LowLevelMonitor.Unix.cs
Просмотреть файл

@ -12,14 +12,10 @@ namespace System.Threading
///
/// Used by the wait subsystem on Unix, so this class cannot have any dependencies on the wait subsystem.
/// </summary>
internal sealed class LowLevelMonitor : IDisposable
internal sealed partial class LowLevelMonitor : IDisposable
{
private IntPtr _nativeMonitor;
#if DEBUG
private RuntimeThread _ownerThread;
#endif
public LowLevelMonitor()
{
_nativeMonitor = Interop.Sys.LowLevelMonitor_New();
@ -27,21 +23,10 @@ namespace System.Threading
{
throw new OutOfMemoryException();
}
#if DEBUG
_ownerThread = null;
#endif
}
~LowLevelMonitor()
private void DisposeCore()
{
Dispose();
}
public void Dispose()
{
VerifyIsNotLockedByAnyThread();
if (_nativeMonitor == IntPtr.Zero)
{
return;
@ -49,96 +34,39 @@ namespace System.Threading
Interop.Sys.LowLevelMonitor_Delete(_nativeMonitor);
_nativeMonitor = IntPtr.Zero;
GC.SuppressFinalize(this);
}
#if DEBUG
public bool IsLocked => _ownerThread == RuntimeThread.CurrentThread;
#endif
public void VerifyIsLocked()
private void AcquireCore()
{
#if DEBUG
Debug.Assert(IsLocked);
#endif
}
public void VerifyIsNotLocked()
{
#if DEBUG
Debug.Assert(!IsLocked);
#endif
}
private void VerifyIsNotLockedByAnyThread()
{
#if DEBUG
Debug.Assert(_ownerThread == null);
#endif
}
private void ResetOwnerThread()
{
#if DEBUG
VerifyIsLocked();
_ownerThread = null;
#endif
}
private void SetOwnerThreadToCurrent()
{
#if DEBUG
VerifyIsNotLockedByAnyThread();
_ownerThread = RuntimeThread.CurrentThread;
#endif
}
public void Acquire()
{
VerifyIsNotLocked();
Interop.Sys.LowLevelMutex_Acquire(_nativeMonitor);
SetOwnerThreadToCurrent();
}
public void Release()
private void ReleaseCore()
{
ResetOwnerThread();
Interop.Sys.LowLevelMutex_Release(_nativeMonitor);
}
public void Wait()
private void WaitCore()
{
ResetOwnerThread();
Interop.Sys.LowLevelMonitor_Wait(_nativeMonitor);
SetOwnerThreadToCurrent();
}
public bool Wait(int timeoutMilliseconds)
private bool WaitCore(int timeoutMilliseconds)
{
Debug.Assert(timeoutMilliseconds >= -1);
if (timeoutMilliseconds < 0)
{
Wait();
WaitCore();
return true;
}
ResetOwnerThread();
bool waitResult = Interop.Sys.LowLevelMonitor_TimedWait(_nativeMonitor, timeoutMilliseconds);
SetOwnerThreadToCurrent();
return waitResult;
return Interop.Sys.LowLevelMonitor_TimedWait(_nativeMonitor, timeoutMilliseconds);
}
public void Signal_Release()
private void Signal_ReleaseCore()
{
ResetOwnerThread();
Interop.Sys.LowLevelMonitor_Signal_Release(_nativeMonitor);
}
/// The following methods typical in a monitor are omitted since they are currently not necessary for the way in which
/// this class is used:
/// - TryAcquire
/// - Signal (use <see cref="Signal_Release"/> instead)
/// - SignalAll
}
}

67
src/System.Private.CoreLib/src/System/Threading/LowLevelMonitor.Windows.cs Normal file
Просмотреть файл

@ -0,0 +1,67 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics;
using System.Runtime.InteropServices;
using Internal.Runtime.Augments;
namespace System.Threading
{
/// <summary>
/// Wraps a critical section and condition variable.
/// </summary>
internal sealed partial class LowLevelMonitor : IDisposable
{
private Interop.Kernel32.CRITICAL_SECTION _criticalSection;
private Interop.Kernel32.CONDITION_VARIABLE _conditionVariable;
public LowLevelMonitor()
{
Interop.Kernel32.InitializeCriticalSection(out _criticalSection);
Interop.Kernel32.InitializeConditionVariable(out _conditionVariable);
}
private void DisposeCore()
{
Interop.Kernel32.DeleteCriticalSection(ref _criticalSection);
}
private void AcquireCore()
{
Interop.Kernel32.EnterCriticalSection(ref _criticalSection);
}
private void ReleaseCore()
{
Interop.Kernel32.LeaveCriticalSection(ref _criticalSection);
}
private void WaitCore()
{
WaitCore(-1);
}
private bool WaitCore(int timeoutMilliseconds)
{
bool waitResult = Interop.Kernel32.SleepConditionVariableCS(ref _conditionVariable, ref _criticalSection, timeoutMilliseconds);
if (!waitResult)
{
int lastError = Marshal.GetLastWin32Error();
if (lastError != Interop.Errors.ERROR_TIMEOUT)
{
var exception = new OutOfMemoryException();
exception.SetErrorCode(lastError);
throw exception;
}
}
return waitResult;
}
private void Signal_ReleaseCore()
{
Interop.Kernel32.WakeConditionVariable(ref _conditionVariable);
Interop.Kernel32.LeaveCriticalSection(ref _criticalSection);
}
}
}

116
src/System.Private.CoreLib/src/System/Threading/LowLevelMonitor.cs Normal file
Просмотреть файл

@ -0,0 +1,116 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics;
using Internal.Runtime.Augments;
namespace System.Threading
{
/// <summary>
/// Wraps a non-recursive mutex and condition.
///
/// Used by the wait subsystem on Unix, so this class cannot have any dependencies on the wait subsystem.
/// </summary>
internal sealed partial class LowLevelMonitor : IDisposable
{
#if DEBUG
private RuntimeThread _ownerThread = null;
#endif
~LowLevelMonitor()
{
Dispose();
}
public void Dispose()
{
VerifyIsNotLockedByAnyThread();
DisposeCore();
GC.SuppressFinalize(this);
}
#if DEBUG
public bool IsLocked => _ownerThread == RuntimeThread.CurrentThread;
#endif
public void VerifyIsLocked()
{
#if DEBUG
Debug.Assert(IsLocked);
#endif
}
public void VerifyIsNotLocked()
{
#if DEBUG
Debug.Assert(!IsLocked);
#endif
}
private void VerifyIsNotLockedByAnyThread()
{
#if DEBUG
Debug.Assert(_ownerThread == null);
#endif
}
private void ResetOwnerThread()
{
#if DEBUG
VerifyIsLocked();
_ownerThread = null;
#endif
}
private void SetOwnerThreadToCurrent()
{
#if DEBUG
VerifyIsNotLockedByAnyThread();
_ownerThread = RuntimeThread.CurrentThread;
#endif
}
public void Acquire()
{
VerifyIsNotLocked();
AcquireCore();
SetOwnerThreadToCurrent();
}
public void Release()
{
ResetOwnerThread();
ReleaseCore();
}
public void Wait()
{
ResetOwnerThread();
WaitCore();
SetOwnerThreadToCurrent();
}
public bool Wait(int timeoutMilliseconds)
{
Debug.Assert(timeoutMilliseconds >= -1);
ResetOwnerThread();
bool waitResult = WaitCore(timeoutMilliseconds);
SetOwnerThreadToCurrent();
return waitResult;
}
public void Signal_Release()
{
ResetOwnerThread();
Signal_ReleaseCore();
}
/// The following methods typical in a monitor are omitted since they are currently not necessary for the way in which
/// this class is used:
/// - TryAcquire
/// - Signal (use <see cref="Signal_Release"/> instead)
/// - SignalAll
}
}

93
src/System.Private.CoreLib/src/System/Threading/ThreadPool.Unix.cs → src/System.Private.CoreLib/src/System/Threading/ThreadPool.Portable.cs
Просмотреть файл

@ -9,7 +9,7 @@ using Microsoft.Win32.SafeHandles;
namespace System.Threading
{
//
// Unix-specific implementation of ThreadPool
// Portable implementation of ThreadPool
//
/// <summary>
@ -325,61 +325,43 @@ namespace System.Threading
public static partial class ThreadPool
{
// TODO: this is a very primitive (temporary) implementation of Thread Pool to allow Tasks to be
// used on Unix. All of this code must be replaced with proper implementation.
/// <summary>
/// Max allowed number of threads in the thread pool. This is just arbitrary number
/// that is used to prevent unbounded creation of threads.
/// It should by high enough to provide sufficient number of thread pool workers
/// in case if some threads get blocked while running user code.
/// </summary>
private static readonly int MaxThreadCount = 4 * ThreadPoolGlobals.processorCount;
/// <summary>
/// Semaphore that is used to release waiting thread pool workers when new work becomes available.
/// </summary>
private static SemaphoreSlim s_semaphore = new SemaphoreSlim(0);
/// <summary>
/// Number of worker threads created by the thread pool.
/// </summary>
private static volatile int s_workerCount = 0;
public static bool SetMaxThreads(int workerThreads, int completionPortThreads)
{
// Not supported at present
return false;
if (workerThreads < 0 || completionPortThreads < 0)
{
return false;
}
return ClrThreadPool.ThreadPoolInstance.SetMaxThreads(workerThreads);
}
public static void GetMaxThreads(out int workerThreads, out int completionPortThreads)
{
// Note that worker threads and completion port threads share the same thread pool.
// The total number of threads cannot exceed MaxThreadCount.
workerThreads = MaxThreadCount;
completionPortThreads = MaxThreadCount;
workerThreads = ClrThreadPool.ThreadPoolInstance.GetMaxThreads();
completionPortThreads = 1;
}
public static bool SetMinThreads(int workerThreads, int completionPortThreads)
{
// Not supported at present
return false;
if (workerThreads < 0 || completionPortThreads < 0)
{
return false;
}
return ClrThreadPool.ThreadPoolInstance.SetMinThreads(workerThreads);
}
public static void GetMinThreads(out int workerThreads, out int completionPortThreads)
{
// All threads are pre-created at present
workerThreads = MaxThreadCount;
completionPortThreads = MaxThreadCount;
workerThreads = ClrThreadPool.ThreadPoolInstance.GetMinThreads();
completionPortThreads = 0;
}
public static void GetAvailableThreads(out int workerThreads, out int completionPortThreads)
{
// Make sure we return a non-negative value if thread pool defaults are changed
int availableThreads = Math.Max(MaxThreadCount - ThreadPoolGlobals.workQueue.numWorkingThreads, 0);
workerThreads = availableThreads;
completionPortThreads = availableThreads;
workerThreads = ClrThreadPool.ThreadPoolInstance.GetAvailableThreads();
completionPortThreads = 0;
}
/// <summary>
@ -387,22 +369,7 @@ namespace System.Threading
/// </summary>
internal static void RequestWorkerThread()
{
// For simplicity of the state management, we pre-create all thread pool workers on the first
// request and then use the semaphore to release threads as new requests come in.
if ((s_workerCount == 0) && Interlocked.Exchange(ref s_workerCount, MaxThreadCount) == 0)
{
for (int i = 0; i < MaxThreadCount; i++)
{
if (!Interop.Sys.RuntimeThread_CreateThread(IntPtr.Zero /*use default stack size*/,
AddrofIntrinsics.AddrOf<Interop.Sys.ThreadProc>(ThreadPoolDispatchCallback), IntPtr.Zero))
{
throw new OutOfMemoryException();
}
}
}
// Release one thread to handle the new request
s_semaphore.Release(1);
ClrThreadPool.ThreadPoolInstance.RequestWorker();
}
internal static bool KeepDispatching(int startTickCount)
@ -412,32 +379,12 @@ namespace System.Threading
internal static void NotifyWorkItemProgress()
{
ClrThreadPool.ThreadPoolInstance.NotifyWorkItemComplete();
}
internal static bool NotifyWorkItemComplete()
{
return true;
}
/// <summary>
/// This method is an entry point of a thread pool worker thread.
/// </summary>
[NativeCallable]
private static IntPtr ThreadPoolDispatchCallback(IntPtr context)
{
var wrapper = ThreadPoolCallbackWrapper.Enter();
do
{
// Handle pending requests
ThreadPoolWorkQueue.Dispatch();
// Wait for new requests to arrive
s_semaphore.Wait();
} while (true);
//wrapper.Exit(resetThread: false);
return ClrThreadPool.ThreadPoolInstance.NotifyWorkItemComplete();
}
private static RegisteredWaitHandle RegisterWaitForSingleObject(

313
tests/src/Simple/Threading/Threading.cs
Просмотреть файл

@ -4,7 +4,9 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
// TODO: Move these tests to CoreFX once they can be run on CoreRT
@ -36,6 +38,32 @@ internal static class Runner
//Console.WriteLine(" WaitSubsystemTests.MutexMaximumReacquireCountTest");
//WaitSubsystemTests.MutexMaximumReacquireCountTest();
Console.WriteLine(" ThreadPoolTests.RunProcessorCountItemsInParallel");
ThreadPoolTests.RunProcessorCountItemsInParallel();
Console.WriteLine(" ThreadPoolTests.RunMoreThanMaxJobsMakesOneJobWaitForStarvationDetection");
ThreadPoolTests.RunMoreThanMaxJobsMakesOneJobWaitForStarvationDetection();
Console.WriteLine(" ThreadPoolTests.ThreadPoolCanPickUpOneJobWhenThreadIsAvailable");
ThreadPoolTests.ThreadPoolCanPickUpOneJobWhenThreadIsAvailable();
Console.WriteLine(" ThreadPoolTests.ThreadPoolCanPickUpMultipleJobsWhenThreadsAreAvailable");
ThreadPoolTests.ThreadPoolCanPickUpMultipleJobsWhenThreadsAreAvailable();
// This test takes a long time to run (min 42 seconds sleeping). Enable for manual testing.
// Console.WriteLine(" ThreadPoolTests.RunJobsAfterThreadTimeout");
// ThreadPoolTests.RunJobsAfterThreadTimeout();
Console.WriteLine(" ThreadPoolTests.WorkQueueDepletionTest");
ThreadPoolTests.WorkQueueDepletionTest();
Console.WriteLine(" ThreadPoolTests.WorkerThreadStateReset");
ThreadPoolTests.WorkerThreadStateReset();
// This test is not applicable (and will not pass) on Windows since it uses the Windows OS-provided thread pool.
// Console.WriteLine(" ThreadPoolTests.SettingMinThreadsWillCreateThreadsUpToMinimum");
// ThreadPoolTests.SettingMinThreadsWillCreateThreadsUpToMinimum();
Console.WriteLine(" WaitThreadTests.SignalingRegisteredHandleCallsCalback");
WaitThreadTests.SignalingRegisteredHandleCallsCalback();
@ -806,6 +834,289 @@ internal static class WaitSubsystemTests
}
}
internal static class ThreadPoolTests
{
[Fact]
public static void RunProcessorCountItemsInParallel()
{
int count = 0;
AutoResetEvent e0 = new AutoResetEvent(false);
for(int i = 0; i < Environment.ProcessorCount; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount)
{
e0.Set();
}
});
}
e0.CheckedWait();
// Run the test again to make sure we can reuse the threads.
count = 0;
for(int i = 0; i < Environment.ProcessorCount; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount)
{
e0.Set();
}
});
}
e0.CheckedWait();
}
[Fact]
public static void RunMoreThanMaxJobsMakesOneJobWaitForStarvationDetection()
{
ManualResetEvent e0 = new ManualResetEvent(false);
AutoResetEvent jobsQueued = new AutoResetEvent(false);
int count = 0;
AutoResetEvent e1 = new AutoResetEvent(false);
for(int i = 0; i < Environment.ProcessorCount; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount)
{
jobsQueued.Set();
}
e0.CheckedWait();
});
}
jobsQueued.CheckedWait();
ThreadPool.QueueUserWorkItem( _ => e1.Set());
Thread.Sleep(500); // Sleep for the gate thread delay to wait for starvation
e1.CheckedWait();
e0.Set();
}
[Fact]
public static void ThreadPoolCanPickUpOneJobWhenThreadIsAvailable()
{
ManualResetEvent e0 = new ManualResetEvent(false);
AutoResetEvent jobsQueued = new AutoResetEvent(false);
AutoResetEvent testJobCompleted = new AutoResetEvent(false);
int count = 0;
for(int i = 0; i < Environment.ProcessorCount - 1; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount - 1)
{
jobsQueued.Set();
}
e0.CheckedWait();
});
}
jobsQueued.CheckedWait();
ThreadPool.QueueUserWorkItem( _ => testJobCompleted.Set());
testJobCompleted.CheckedWait();
e0.Set();
}
[Fact]
public static void ThreadPoolCanPickUpMultipleJobsWhenThreadsAreAvailable()
{
ManualResetEvent e0 = new ManualResetEvent(false);
AutoResetEvent jobsQueued = new AutoResetEvent(false);
AutoResetEvent testJobCompleted = new AutoResetEvent(false);
int count = 0;
for(int i = 0; i < Environment.ProcessorCount - 1; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount - 1)
{
jobsQueued.Set();
}
e0.CheckedWait();
});
}
jobsQueued.CheckedWait();
int testJobsCount = 0;
int maxCount = 5;
void Job(object _)
{
if(Interlocked.Increment(ref testJobsCount) != maxCount)
{
ThreadPool.QueueUserWorkItem(Job);
}
else
{
testJobCompleted.Set();
}
}
ThreadPool.QueueUserWorkItem(Job);
testJobCompleted.CheckedWait();
e0.Set();
}
private static WaitCallback CreateRecursiveJob(int jobCount, int targetJobCount, AutoResetEvent testJobCompleted)
{
return _ =>
{
if (jobCount == targetJobCount)
{
testJobCompleted.Set();
}
else
{
ThreadPool.QueueUserWorkItem(CreateRecursiveJob(jobCount + 1, targetJobCount, testJobCompleted));
}
};
}
[Fact]
[OuterLoop]
public static void RunJobsAfterThreadTimeout()
{
ManualResetEvent e0 = new ManualResetEvent(false);
AutoResetEvent jobsQueued = new AutoResetEvent(false);
AutoResetEvent testJobCompleted = new AutoResetEvent(false);
int count = 0;
for(int i = 0; i < Environment.ProcessorCount - 1; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == Environment.ProcessorCount - 1)
{
jobsQueued.Set();
}
e0.CheckedWait();
});
}
jobsQueued.CheckedWait();
ThreadPool.QueueUserWorkItem( _ => testJobCompleted.Set());
testJobCompleted.CheckedWait();
Console.Write("Sleeping to time out thread\n");
Thread.Sleep(21000);
ThreadPool.QueueUserWorkItem( _ => testJobCompleted.Set());
testJobCompleted.CheckedWait();
e0.Set();
Console.Write("Sleeping to time out all threads\n");
Thread.Sleep(21000);
ThreadPool.QueueUserWorkItem( _ => testJobCompleted.Set());
testJobCompleted.CheckedWait();
}
[Fact]
public static void WorkQueueDepletionTest()
{
ManualResetEvent e0 = new ManualResetEvent(false);
int numLocalScheduled = 1;
int numGlobalScheduled = 1;
int numToSchedule = Environment.ProcessorCount * 64;
int numCompleted = 0;
object syncRoot = new object();
void ThreadLocalJob()
{
if(Interlocked.Increment(ref numLocalScheduled) <= numToSchedule)
{
Task.Factory.StartNew(ThreadLocalJob);
}
if(Interlocked.Increment(ref numLocalScheduled) <= numToSchedule)
{
Task.Factory.StartNew(ThreadLocalJob);
}
if (Interlocked.Increment(ref numCompleted) == numToSchedule * 2)
{
e0.Set();
}
}
void GlobalJob(object _)
{
if(Interlocked.Increment(ref numGlobalScheduled) <= numToSchedule)
{
ThreadPool.QueueUserWorkItem(GlobalJob);
}
if(Interlocked.Increment(ref numGlobalScheduled) <= numToSchedule)
{
ThreadPool.QueueUserWorkItem(GlobalJob);
}
if (Interlocked.Increment(ref numCompleted) == numToSchedule * 2)
{
e0.Set();
}
}
Task.Factory.StartNew(ThreadLocalJob);
ThreadPool.QueueUserWorkItem(GlobalJob);
e0.CheckedWait();
}
[Fact]
public static void WorkerThreadStateReset()
{
var cultureInfo = new CultureInfo("pt-BR");
var expectedCultureInfo = CultureInfo.CurrentCulture;
var expectedUICultureInfo = CultureInfo.CurrentUICulture;
int count = 0;
AutoResetEvent e0 = new AutoResetEvent(false);
for(int i = 0; i < Environment.ProcessorCount; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
CultureInfo.CurrentCulture = cultureInfo;
CultureInfo.CurrentUICulture = cultureInfo;
Thread.CurrentThread.Priority = ThreadPriority.AboveNormal;
if(Interlocked.Increment(ref count) == Environment.ProcessorCount)
{
e0.Set();
}
});
}
e0.CheckedWait();
// Run the test again to make sure we can reuse the threads.
count = 0;
for(int i = 0; i < Environment.ProcessorCount; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
Assert.Equal(expectedCultureInfo, CultureInfo.CurrentCulture);
Assert.Equal(expectedUICultureInfo, CultureInfo.CurrentUICulture);
Assert.Equal(ThreadPriority.Normal, Thread.CurrentThread.Priority);
if(Interlocked.Increment(ref count) == Environment.ProcessorCount)
{
e0.Set();
}
});
}
e0.CheckedWait();
}
[Fact]
public static void SettingMinThreadsWillCreateThreadsUpToMinimum()
{
ThreadPool.GetMinThreads(out int minThreads, out int unusedMin);
ThreadPool.GetMaxThreads(out int maxThreads, out int unusedMax);
try
{
ManualResetEvent e0 = new ManualResetEvent(false);
AutoResetEvent jobsQueued = new AutoResetEvent(false);
int count = 0;
ThreadPool.SetMaxThreads(minThreads, unusedMax);
for(int i = 0; i < minThreads + 1; ++i)
{
ThreadPool.QueueUserWorkItem( _ => {
if(Interlocked.Increment(ref count) == minThreads + 1)
{
jobsQueued.Set();
}
e0.CheckedWait();
});
}
Assert.False(jobsQueued.WaitOne(ThreadTestHelpers.ExpectedTimeoutMilliseconds));
Assert.True(ThreadPool.SetMaxThreads(minThreads + 1, unusedMax));
Assert.True(ThreadPool.SetMinThreads(minThreads + 1, unusedMin));
jobsQueued.CheckedWait();
e0.Set();
}
finally
{
ThreadPool.SetMinThreads(minThreads, unusedMin);
ThreadPool.SetMaxThreads(maxThreads, unusedMax);
}
}
}
internal static class WaitThreadTests
{
private const int WaitThreadTimeoutTimeMs = 20000;
@ -1017,8 +1328,6 @@ internal static class ThreadTestHelpers
}
}
internal sealed class InvalidWaitHandle : WaitHandle
{
}