Pourquoi Monitor.PulseAll génère-t-il un modèle de latence «d’escalier» dans les threads signalés?

Dans une bibliothèque utilisant Monitor.PulseAll () pour la synchronisation des threads, j’ai remarqué que la latence entre le moment où PulseAll (…) est appelé et le moment où un thread est réveillé semble suivre une dissortingbution “stepping stair” – avec grandes marches. Les fils réveillés ne font presque pas de travail; et presque immédiatement à attendre sur le moniteur. Par exemple, sur une boîte de 12 cœurs avec 24 threads en attente sur un moniteur (2x Xeon5680 / Gulftown; 6 cœurs physiques par processeur; HT désactivé), la latence entre le réveil et le thread est telle que:

Les 12 premiers threads (notez que nous avons 12 cœurs) prennent entre 30 et 60 microsecondes pour répondre. Ensuite, nous commençons à obtenir de très gros sauts; avec les plateaux autour de 700, 1300, 1900 et 2600 microsecondes.

J’ai réussi à recréer ce comportement indépendamment de la bibliothèque tierce en utilisant le code ci-dessous. Ce que fait ce code est de lancer un grand nombre de threads (changer le paramètre numThreads) qui attend juste sur un moniteur, lit un horodatage, le connecte à un ConcurrentSet, puis retourne immédiatement à Waiting. Une fois par seconde, PulseAll () réveille tous les threads. Il le fait 20 fois et rapporte les latences pour la 10ème itération à la console.

• Quel est le problème résolu par IStructuralEquatable et IStructuralComparable?
• Contrôle WPF WebBrowser – Comment supprimer les erreurs de script?
• Supprimer l’élément de la liste en fonction de la condition
• Devriez-vous implémenter IDisposable.Dispose () de sorte qu’il ne jette jamais?
• LINQ – Syntaxe de requête vs chaîne de méthode & lambda

using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; using System.Collections.Concurrent; using System.Diagnostics; namespace PulseAllTest { class Program { static long LastTimestamp; static long Iteration; static object SyncObj = new object(); static Stopwatch s = new Stopwatch(); static ConcurrentBag<Tuple> IterationToTicks = new ConcurrentBag<Tuple>(); static void Main(ssortingng[] args) { long numThreads = 32; for (int i = 0; i < numThreads; ++i) { Task.Factory.StartNew(ReadLastTimestampAndPublish, TaskCreationOptions.LongRunning); } s.Start(); for (int i = 0; i < 20; ++i) { lock (SyncObj) { ++Iteration; LastTimestamp = s.Elapsed.Ticks; Monitor.PulseAll(SyncObj); } Thread.Sleep(TimeSpan.FromSeconds(1)); } Console.WriteLine(String.Join("\n", from n in IterationToTicks where n.Item1 == 10 orderby n.Item2 select ((decimal)n.Item2)/TimeSpan.TicksPerMillisecond)); Console.Read(); } static void ReadLastTimestampAndPublish() { while(true) { lock(SyncObj) { Monitor.Wait(SyncObj); } IterationToTicks.Add(Tuple.Create(Iteration, s.Elapsed.Ticks - LastTimestamp)); } } } } 

En utilisant le code ci-dessus, voici un exemple de latences sur une boîte avec 8 cœurs / w hyperthreading activé (soit 16 cœurs dans le Gestionnaire des tâches) et 32 ​​threads (* 2x Xeon5550 / Gainestown; 4 cœurs physiques par processeur; HT activé):

EDIT: Pour essayer de prendre NUMA hors de l’équation, voici un graphique exécutant le programme exemple avec 16 threads sur un Core i7-3770 (Ivy Bridge); 4 kernelx physiques; HT activé:

Quelqu’un peut-il expliquer pourquoi Monitor.PulseAll () se comporte de cette façon?

EDIT2:

Pour essayer de montrer que ce comportement n’est pas inhérent à l’éveil simultané d’un tas de threads, j’ai répliqué le comportement du programme de test en utilisant Events; et au lieu de mesurer la latence de PulseAll (), je mesure la latence de ManualResetEvent.Set (). Le code crée un certain nombre de threads de travail, puis attend un événement ManualResetEvent.Set () sur le même object ManualResetEvent. Lorsque l’événement est déclenché, ils prennent une mesure de latence puis attendent immédiatement leur propre AutoResetEvent par thread. Bien avant la prochaine itération (500 ms avant), ManualResetEvent est réinitialisé () et chaque AutoResetEvent est défini sur Set () pour que les threads puissent attendre à nouveau sur le ManualResetEvent partagé.

J’ai hésité à poster ceci parce que cela pourrait être une audition rouge géante (je ne prétends pas que les événements et les moniteurs se comportent de la même manière) et qu’il utilise des pratiques absolument terribles pour qu’un événement se comporte comme un moniteur (j’aimerais les collègues feraient si je soumettais ceci à une revue de code); mais je pense que les résultats sont éclairants.

Ce test a été effectué sur la même machine que le test d’origine. un 2xXeon5680 / Gulftown; 6 cœurs par processeur (12 cœurs au total); Hyperthreading désactivé.

Si ce n’est pas évident, c’est radicalement différent de Monitor.PulseAll; voici le premier graphique superposé sur le dernier graphique:

Le code utilisé pour générer ces mesures est ci-dessous:

 using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; using System.Collections.Concurrent; using System.Diagnostics; namespace MRETest { class Program { static long LastTimestamp; static long Iteration; static ManualResetEventSlim MRES = new ManualResetEventSlim(false); static List Publishers = new List(); static Stopwatch s = new Stopwatch(); static ConcurrentBag<Tuple> IterationToTicks = new ConcurrentBag<Tuple>(); static void Main(ssortingng[] args) { long numThreads = 24; s.Start(); for (int i = 0; i < numThreads; ++i) { AutoResetEvent ares = new AutoResetEvent(false); ReadLastTimestampAndPublish spinner = new ReadLastTimestampAndPublish( new AutoResetEvent(false)); Task.Factory.StartNew(spinner.Spin, TaskCreationOptions.LongRunning); Publishers.Add(spinner); } for (int i = 0; i < 20; ++i) { ++Iteration; LastTimestamp = s.Elapsed.Ticks; MRES.Set(); Thread.Sleep(500); MRES.Reset(); foreach (ReadLastTimestampAndPublish publisher in Publishers) { publisher.ARES.Set(); } Thread.Sleep(500); } Console.WriteLine(String.Join("\n", from n in IterationToTicks where n.Item1 == 10 orderby n.Item2 select ((decimal)n.Item2) / TimeSpan.TicksPerMillisecond)); Console.Read(); } class ReadLastTimestampAndPublish { public AutoResetEvent ARES { get; private set; } public ReadLastTimestampAndPublish(AutoResetEvent ares) { this.ARES = ares; } public void Spin() { while (true) { MRES.Wait(); IterationToTicks.Add(Tuple.Create(Iteration, s.Elapsed.Ticks - LastTimestamp)); ARES.WaitOne(); } } } } } 

• Comment faire des tests d’intégration dans .NET avec de vrais fichiers?
• Utilisation du numéro de version de l’application Web à partir d’un assembly (ASP.NET/C#)
• Pourquoi ne sums-nous pas autorisés à spécifier un constructeur dans une interface?
• Masquer le formulaire au lieu de le fermer lorsque le bouton de fermeture est cliqué
• Comment lire correctement un champ Interlocked.Increment’ed int?

 static void ReadLastTimestampAndPublish() { while(true) { lock(SyncObj) { Monitor.Wait(SyncObj); } IterationToTicks.Add(Tuple.Create(Iteration, s.Elapsed.Ticks - LastTimestamp)); Thread.Sleep(TimeSpan.FromMilliseconds(100)); // <=== } } 

 namespace System.Threading { public static class Monitor { // other methods omitted [MethodImplAtsortingbute(MethodImplOptions.InternalCall)] private static extern void ObjPulseAll(Object obj); public static void PulseAll(Object obj) { if (obj==null) { throw new ArgumentNullException("obj"); } ObjPulseAll(obj); } } } 

 FCFuncStart(gMonitorFuncs) FCFuncElement("Enter", JIT_MonEnter) FCFuncElement("Exit", JIT_MonExit) FCFuncElement("TryEnterTimeout", JIT_MonTryEnter) FCFuncElement("ObjWait", ObjectNative::WaitTimeout) FCFuncElement("ObjPulse", ObjectNative::Pulse) FCFuncElement("ObjPulseAll", ObjectNative::PulseAll) FCFuncElement("ReliableEnter", JIT_MonReliableEnter) FCFuncEnd() 

 FCIMPL1(void, ObjectNative::PulseAll, Object* pThisUNSAFE) { CONTRACTL { MODE_COOPERATIVE; DISABLED(GC_TRIGGERS); // can't use this in an FCALL because we're in forbid gc mode until we setup a H_M_F. THROWS; SO_TOLERANT; } CONTRACTL_END; OBJECTREF pThis = (OBJECTREF) pThisUNSAFE; HELPER_METHOD_FRAME_BEGIN_1(pThis); //-[autocvtpro]------------------------------------------------------- if (pThis == NULL) COMPlusThrow(kNullReferenceException, L"NullReference_This"); pThis->PulseAll(); //-[autocvtepi]------------------------------------------------------- HELPER_METHOD_FRAME_END(); } FCIMPLEND 

 class Object { // snip public: // snip ObjHeader *GetHeader() { LEAF_CONTRACT; return PTR_ObjHeader(PTR_HOST_TO_TADDR(this) - sizeof(ObjHeader)); } // snip void PulseAll() { WRAPPER_CONTRACT; GetHeader()->PulseAll(); } // snip } 

 void ObjHeader::PulseAll() { CONTRACTL { INSTANCE_CHECK; THROWS; GC_TRIGGERS; MODE_ANY; INJECT_FAULT(COMPlusThrowOM();); } CONTRACTL_END; // The following code may cause GC, so we must fetch the sync block from // the object now in case it moves. SyncBlock *pSB = GetBaseObject()->GetSyncBlock(); // GetSyncBlock throws on failure _ASSERTE(pSB != NULL); // make sure we own the crst if (!pSB->DoesCurrentThreadOwnMonitor()) COMPlusThrow(kSynchronizationLockException); pSB->PulseAll(); } void SyncBlock::PulseAll() { CONTRACTL { INSTANCE_CHECK; NOTHROW; GC_NOTRIGGER; MODE_ANY; } CONTRACTL_END; WaitEventLink *pWaitEventLink; while ((pWaitEventLink = ThreadQueue::DequeueThread(this)) != NULL) pWaitEventLink->m_EventWait->Set(); }