Associating threadpool queues with CPU cores.

src/System.Private.CoreLib/shared/System/Threading/Tasks/ThreadPoolTaskScheduler.cs

@@ -69,8 +69,16 @@ protected internal override void QueueTask(Task task)
         protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
         {
             // If the task was previously scheduled, and we can't pop it, then return false.
-            if (taskWasPreviouslyQueued && !ThreadPool.TryPopCustomWorkItem(task))
-                return false;
+            if (taskWasPreviouslyQueued)
+            {
+                // do not inline in a nontrivial sync context - it could be stricter than what enqueuer had.
+                SynchronizationContext? syncCtx = SynchronizationContext.Current;
+                if (syncCtx != null && syncCtx.GetType() != typeof(SynchronizationContext))
+                    return false;
+
+                if (!ThreadPool.TryPopCustomWorkItem(task))
+                    return false;
+            }
 
             try
             {

src/System.Private.CoreLib/src/System/Threading/Thread.CoreCLR.cs

@@ -519,6 +519,14 @@ private static int RefreshCurrentProcessorId()
             return currentProcessorId;
         }
 
+        // Clear the cached processor Id
+        // This can be used before/after blocking the thread for nontrivial amount of time
+        // or around other operations which are likely to result in changing executing core.
+        internal static void FlushCurrentProcessorId()
+        {
+            t_currentProcessorIdCache = default;
+        }
+
         // Cached processor id used as a hint for which per-core stack to access. It is periodically
         // refreshed to trail the actual thread core affinity.
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

src/vm/threadpoolrequest.cpp

@@ -537,18 +537,9 @@ void ManagedPerAppDomainTPCount::SetAppDomainRequestsActive()
     _ASSERTE(m_index.m_dwIndex != UNUSED_THREADPOOL_INDEX);
 
 #ifndef DACCESS_COMPILE
-        LONG count = VolatileLoad(&m_numRequestsPending);
-        while (true)
-        {
-            LONG prev = FastInterlockCompareExchange(&m_numRequestsPending, count+1, count);
-            if (prev == count)
-            {
-                ThreadpoolMgr::MaybeAddWorkingWorker();
-                ThreadpoolMgr::EnsureGateThreadRunning();
-                break;
-            }
-            count = prev;
-        }
+    FastInterlockIncrement(&m_numRequestsPending);
+    ThreadpoolMgr::MaybeAddWorkingWorker();
+    ThreadpoolMgr::EnsureGateThreadRunning();
 #endif
 }
 
@@ -560,20 +551,13 @@ void ManagedPerAppDomainTPCount::ClearAppDomainRequestsActive()
 
     _ASSERTE(m_index.m_dwIndex != UNUSED_THREADPOOL_INDEX);
 
-    LONG count = VolatileLoad(&m_numRequestsPending);
-    while (count > 0)
-    {
-        LONG prev = FastInterlockCompareExchange(&m_numRequestsPending, 0, count);
-        if (prev == count)
-            break;
-        count = prev;
-    }
+    m_numRequestsPending = 0;
 }
 
 bool ManagedPerAppDomainTPCount::TakeActiveRequest()
 {
     LIMITED_METHOD_CONTRACT;
-    LONG count = VolatileLoad(&m_numRequestsPending);
+    LONG count = VolatileLoadWithoutBarrier(&m_numRequestsPending);
     while (count > 0)
     {
         LONG prev = FastInterlockCompareExchange(&m_numRequestsPending, count-1, count);

src/vm/threadpoolrequest.h

@@ -135,11 +135,9 @@ class ManagedPerAppDomainTPCount : public IPerAppDomainTPCount {
 
 private:
     TPIndex m_index;
-    struct DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) {
-        BYTE m_padding1[MAX_CACHE_LINE_SIZE - sizeof(LONG)];
+    struct ALIGNED(MAX_CACHE_LINE_SIZE) {
         // Only use with VolatileLoad+VolatileStore+FastInterlockCompareExchange
         LONG m_numRequestsPending;
-        BYTE m_padding2[MAX_CACHE_LINE_SIZE];
     };
 };
 
@@ -230,11 +228,9 @@ class UnManagedPerAppDomainTPCount : public IPerAppDomainTPCount {
 private:
     SpinLock m_lock;
     ULONG m_NumRequests;
-    struct DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) {
-        BYTE m_padding1[MAX_CACHE_LINE_SIZE - sizeof(LONG)];
+    struct ALIGNED(MAX_CACHE_LINE_SIZE) {
         // Only use with VolatileLoad+VolatileStore+FastInterlockCompareExchange
         LONG m_outstandingThreadRequestCount;
-        BYTE m_padding2[MAX_CACHE_LINE_SIZE];
     };
 };
 

src/vm/util.hpp

@@ -48,6 +48,12 @@
 #define MAX_CACHE_LINE_SIZE 64
 #endif
 
+#ifdef _MSC_VER
+#define ALIGNED(x)   __declspec(align(x))
+#else
+#define ALIGNED(x)   __attribute__((aligned(x)))
+#endif
+
 //========================================================================
 // More convenient names for integer types of a guaranteed size.
 //========================================================================

src/vm/win32threadpool.cpp

@@ -77,6 +77,7 @@ SetWaitableTimerExProc g_pufnSetWaitableTimerEx = NULL;
 BOOL ThreadpoolMgr::InitCompletionPortThreadpool = FALSE;
 HANDLE ThreadpoolMgr::GlobalCompletionPort;                 // used for binding io completions on file handles
 
+// Cacheline aligned
 SVAL_IMPL(ThreadpoolMgr::ThreadCounter,ThreadpoolMgr,CPThreadCounter);
 
 SVAL_IMPL_INIT(LONG,ThreadpoolMgr,MaxLimitTotalCPThreads,1000);   // = MaxLimitCPThreadsPerCPU * number of CPUS
@@ -85,8 +86,8 @@ SVAL_IMPL(LONG,ThreadpoolMgr,MaxFreeCPThreads);                   // = MaxFreeCP
 
 Volatile<LONG> ThreadpoolMgr::NumCPInfrastructureThreads = 0;      // number of threads currently busy handling draining cycle
 
-// Cacheline aligned, hot variable
-DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) SVAL_IMPL(ThreadpoolMgr::ThreadCounter, ThreadpoolMgr, WorkerCounter);
+// Cacheline aligned
+SVAL_IMPL(ThreadpoolMgr::ThreadCounter, ThreadpoolMgr, WorkerCounter);
 
 SVAL_IMPL(LONG,ThreadpoolMgr,MinLimitTotalWorkerThreads);          // = MaxLimitCPThreadsPerCPU * number of CPUS
 SVAL_IMPL(LONG,ThreadpoolMgr,MaxLimitTotalWorkerThreads);        // = MaxLimitCPThreadsPerCPU * number of CPUS
@@ -97,12 +98,9 @@ LONG    ThreadpoolMgr::cpuUtilizationAverage = 0;
 HillClimbing ThreadpoolMgr::HillClimbingInstance;
 
 // Cacheline aligned, 3 hot variables updated in a group
-DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) LONG ThreadpoolMgr::PriorCompletedWorkRequests = 0;
-DWORD ThreadpoolMgr::PriorCompletedWorkRequestsTime;
-DWORD ThreadpoolMgr::NextCompletedWorkRequestsTime;
+ThreadpoolMgr::WorkRequestDataT ThreadpoolMgr::WorkRequestData;
 
 LARGE_INTEGER ThreadpoolMgr::CurrentSampleStartTime;
-
 unsigned int ThreadpoolMgr::WorkerThreadSpinLimit;
 bool ThreadpoolMgr::IsHillClimbingDisabled;
 int ThreadpoolMgr::ThreadAdjustmentInterval;
@@ -924,7 +922,7 @@ void ThreadpoolMgr::AdjustMaxWorkersActive()
 
     DWORD currentTicks = GetTickCount();
     LONG totalNumCompletions = (LONG)Thread::GetTotalWorkerThreadPoolCompletionCount();
-    LONG numCompletions = totalNumCompletions - VolatileLoad(&PriorCompletedWorkRequests);
+    LONG numCompletions = totalNumCompletions - VolatileLoad(&WorkRequestData.PriorCompletedWorkRequests);
 
     LARGE_INTEGER startTime = CurrentSampleStartTime;
     LARGE_INTEGER endTime;
@@ -986,10 +984,12 @@ void ThreadpoolMgr::AdjustMaxWorkersActive()
             }
         }
 
-        PriorCompletedWorkRequests = totalNumCompletions;
-        NextCompletedWorkRequestsTime = currentTicks + ThreadAdjustmentInterval;
-        MemoryBarrier(); // flush previous writes (especially NextCompletedWorkRequestsTime)
-        PriorCompletedWorkRequestsTime = currentTicks;
+        WorkRequestData.PriorCompletedWorkRequests = totalNumCompletions;
+        WorkRequestData.NextCompletedWorkRequestsTime = currentTicks + ThreadAdjustmentInterval;
+        // make sure that NextCompletedWorkRequestsTime is updated before PriorCompletedWorkRequestsTime
+        // so that reader never sees newer PriorCompletedWorkRequestsTime while having stale NextCompletedWorkRequestsTime
+        // NB: we are holding the ThreadAdjustmentLock and therefore the order cannot be violated by two threads writing.
+        VolatileStore(&WorkRequestData.PriorCompletedWorkRequestsTime, currentTicks);
         CurrentSampleStartTime = endTime;;
     }
 }
@@ -1011,13 +1011,14 @@ void ThreadpoolMgr::MaybeAddWorkingWorker()
     while (true)
     {
         newCounts = counts;
-        newCounts.NumWorking = max(counts.NumWorking, min(counts.NumWorking + 1, counts.MaxWorking));
-        newCounts.NumActive = max(counts.NumActive, newCounts.NumWorking);
-        newCounts.NumRetired = max(0, counts.NumRetired - (newCounts.NumActive - counts.NumActive));
 
-        if (newCounts == counts)
+        if (counts.NumWorking + 1 > counts.MaxWorking)
             return;
 
+        newCounts.NumWorking++;
+        newCounts.NumActive = max(counts.NumActive, newCounts.NumWorking);
+        newCounts.NumRetired = max(0, counts.NumRetired - (newCounts.NumActive - counts.NumActive));
+
         ThreadCounter::Counts oldCounts = WorkerCounter.CompareExchangeCounts(newCounts, counts);
 
         if (oldCounts == counts)

src/vm/win32threadpool.h

@@ -106,7 +106,9 @@ class ThreadpoolMgr
     friend struct _DacGlobals;
 
 public:
-    struct ThreadCounter
+
+    // ThreadCounter is always cacheline aligned
+    struct ALIGNED(MAX_CACHE_LINE_SIZE) ThreadCounter
     {
         static const int MaxPossibleCount = 0x7fff;
 
@@ -299,7 +301,7 @@ class ThreadpoolMgr
     static inline void UpdateLastDequeueTime()
     {
         LIMITED_METHOD_CONTRACT;
-        VolatileStore(&LastDequeueTime, (unsigned int)GetTickCount());
+        LastDequeueTime = (unsigned int)GetTickCount();
     }
 
     static BOOL CreateTimerQueueTimer(PHANDLE phNewTimer,
@@ -851,9 +853,11 @@ class ThreadpoolMgr
     {
         WRAPPER_NO_CONTRACT;
 
-        DWORD priorTime = PriorCompletedWorkRequestsTime;
-        MemoryBarrier(); // read fresh value for NextCompletedWorkRequestsTime below
-        DWORD requiredInterval = NextCompletedWorkRequestsTime - priorTime;
+        // make sure that PriorCompletedWorkRequestsTime is read before NextCompletedWorkRequestsTime
+        // to make sure that NextCompletedWorkRequestsTime is not older than PriorCompletedWorkRequestsTime
+        // NB: we write them in reverse order while holding a lock.
+        DWORD priorTime = VolatileLoad(&WorkRequestData.PriorCompletedWorkRequestsTime);
+        DWORD requiredInterval = WorkRequestData.NextCompletedWorkRequestsTime - priorTime;
         DWORD elapsedInterval = GetTickCount() - priorTime;
         if (elapsedInterval >= requiredInterval)
         {
@@ -1016,13 +1020,21 @@ class ThreadpoolMgr
 
     static HillClimbing HillClimbingInstance;
 
-    DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) static LONG PriorCompletedWorkRequests;
-    static DWORD PriorCompletedWorkRequestsTime;
-    static DWORD NextCompletedWorkRequestsTime;
+    // the following 3 hot variables are nearly always used together. 
+    // put on a separate cache line.
+    struct ALIGNED(MAX_CACHE_LINE_SIZE) WorkRequestDataT
+    {
+    public:
+        LONG PriorCompletedWorkRequests;
+        DWORD PriorCompletedWorkRequestsTime;
+        DWORD NextCompletedWorkRequestsTime;
+    };
+
+    static WorkRequestDataT WorkRequestData;
 
     static LARGE_INTEGER CurrentSampleStartTime;
 
-    static unsigned int WorkerThreadSpinLimit;
+    static unsigned int ThreadpoolMgr::WorkerThreadSpinLimit;
     static bool IsHillClimbingDisabled;
     static int ThreadAdjustmentInterval;
 
@@ -1044,7 +1056,7 @@ class ThreadpoolMgr
     static const DWORD WorkerTimeout = 20 * 1000;
     static const DWORD WorkerTimeoutAppX = 5 * 1000;    // shorter timeout to allow threads to exit prior to app suspension
 
-    DECLSPEC_ALIGN(MAX_CACHE_LINE_SIZE) SVAL_DECL(ThreadCounter,WorkerCounter);
+    SVAL_DECL(ThreadCounter, WorkerCounter);
 
     // 
     // WorkerSemaphore is an UnfairSemaphore because: