Время межпотоковой связи

Question

Я соединяю 15 асинхронных операций через порты и приемники. Это меня очень беспокоило время межсетевого обмена сообщениями, в частности время, затрачиваемое на перенос данных задачи на порт, и новая задача начинает обрабатывать эти же данные в другом потоке. Предполагая, что в начале каждого случая каждый поток неактивен, я создал тест, который использует класс секундомера для измерения времени от двух разных диспетчеров, каждый из которых работает с наивысшим приоритетом одним потоком.

Что меня удивило, моя установка для разработки - это Q6600 Quad Core 2.4 Ghz, работающий под управлением Windows 7 x64, а среднее время переключения контекста из моего теста составляло 5,66 микросекунды со стандартным отклонением 5,738 микросекунды и максимум почти 1,58 миллисекунды (в 282 раза!). Частота секундомера составляет 427,7 наносекунды, поэтому я все еще не чувствую шума датчика.

Что бы я хотел сделать, это сократить максимально возможное время обмена сообщениями и, что не менее важно, уменьшить стандартное отклонение контекстного переключателя. Я понимаю, что Windows не является оперативной оперативной системой, и нет гарантий, но планировщик окон - это график с честным круговым распределением по приоритетам, и два потока в этом тесте оба имеют наивысший приоритет (единственные потоки, которые должны быть такими high), поэтому не должно быть каких-либо контекстных переключателей на потоках (очевидно, что максимальное время на 1,58 мс ... Я считаю, что кванты окон составляют 15,65 мс?) Единственное, о чем я могу думать, это изменение сроков вызовов ОС к механизмам блокировки, используемым CCR для передачи сообщений между потоками.

Пожалуйста, дайте мне знать, если кто-то еще измерил время обмена сообщениями, и есть предложения по его улучшению.

Вот исходный код из моих тестов:

using System; 
using System.Collections.Generic; 
using System.IO; 
using System.Threading; 
using Microsoft.Ccr.Core; 

using System.Diagnostics; 

namespace Test.CCR.TestConsole 
{ 
    class Program 
    { 
     static void Main(string[] args) 
     { 
      Console.WriteLine("Starting Timer"); 
      var sw = new Stopwatch(); 
      sw.Start(); 

      var dispatcher = new Dispatcher(1, ThreadPriority.Highest, true, "My Thread Pool"); 
      var dispQueue = new DispatcherQueue("Disp Queue", dispatcher); 

      var sDispatcher = new Dispatcher(1, ThreadPriority.Highest, true, "Second Dispatcher"); 
      var sDispQueue = new DispatcherQueue("Second Queue", sDispatcher); 

      var legAPort = new Port<EmptyValue>(); 
      var legBPort = new Port<TimeSpan>(); 

      var distances = new List<double>(); 

      long totalTicks = 0; 

      while (sw.Elapsed.TotalMilliseconds < 5000) ; 

      int runCnt = 100000; 
      int offset = 1000; 

      Arbiter.Activate(dispQueue, Arbiter.Receive(true, legAPort, i => 
                      { 
                       TimeSpan sTime = sw.Elapsed; 
                       legBPort.Post(sTime); 
                      })); 
      Arbiter.Activate(sDispQueue, Arbiter.Receive(true, legBPort, i => 
                      { 
                       TimeSpan eTime = sw.Elapsed; 
                       TimeSpan dt = eTime.Subtract(i); 
                       //if (distances.Count == 0 || Math.Abs(distances[distances.Count - 1] - dt.TotalMilliseconds)/distances[distances.Count - 1] > 0.1) 
                       distances.Add(dt.TotalMilliseconds); 

                       if(distances.Count > offset) 
                       Interlocked.Add(ref totalTicks, 
                           dt.Ticks); 
                       if(distances.Count < runCnt) 
                        legAPort.Post(EmptyValue.SharedInstance); 
                      })); 


      //Thread.Sleep(100); 
      legAPort.Post(EmptyValue.SharedInstance); 

      Thread.Sleep(500); 

      while (distances.Count < runCnt) 
       Thread.Sleep(25); 

      TimeSpan exTime = TimeSpan.FromTicks(totalTicks); 
      double exMS = exTime.TotalMilliseconds/(runCnt - offset); 

      Console.WriteLine("Exchange Time: {0} Stopwatch Resolution: {1}", exMS, Stopwatch.Frequency); 

      using(var stw = new StreamWriter("test.csv")) 
      { 
       for(int ix=0; ix < distances.Count; ix++) 
       { 
        stw.WriteLine("{0},{1}", ix, distances[ix]); 
       } 
       stw.Flush(); 
      } 

      Console.ReadKey(); 
     } 
    } 
} 

Answer 1

Windows не является оперативной системой. Но вы уже это знали. То, что убивает вас, - это время переключения контекста, а не время сообщения. Вы действительно не указали, КАК ваша работа между процессами работает.Если вы действительно используете несколько потоков, вы можете получить некоторую выгоду, не используя сообщение Windows в качестве протокола связи, вместо этого попробуйте скопировать свой собственный IPC, используя вместо него очереди сообщений, размещенных в приложении.

Лучшее среднее значение, на которое вы можете рассчитывать, составляет 1 мс с любой версией Windows при переключении контекста. Вероятно, вы видели 1 миллион раз, когда ваше приложение должно уступить ядру. Это по дизайну для приложений Ring-1 (пользовательское пространство). Если абсолютно важно, чтобы вы опустились ниже 1 мс, вам нужно переключить часть вашего приложения на Ring-0, что означает запись драйвера устройства.

Драйверы устройств не страдают от того же времени переключения контекста, что и у пользовательских приложений, и имеют доступ к таймерам с разрешением nano-second и вызовам сна. Если вам это нужно, DDK (комплект разработчика драйверов устройств) можно бесплатно получить у Microsoft, но я бы настоятельно рекомендовал вам инвестировать в сторонний комплект разработчика. У них обычно есть действительно хорошие образцы и множество волшебников, чтобы правильно настроить вещи, которые заставят вас месяцами читать документы DDK, чтобы их обнаружить. Вы также захотите получить что-то вроде SoftIce, потому что обычный отладчик Visual Studio не поможет вам отлаживать драйверы устройств.

Answer 2

ThreadPriority.Highest не означает, что только нить сам планировщик имеет более высокий приоритет. У Win32 API есть более узкий уровень приоритета потока (clicky) с несколькими уровнями выше самого высокого (по крайней мере, IIRC Highest - это самый высокий приоритет, который не может быть введен администратором, администраторы могут планировать более высокий приоритет, как и любые аппаратные драйверы/код режима ядра), поэтому нет никакой гарантии, что они не будут упреждаться.

Даже если поток работает с окнами с наивысшим приоритетом, он может продвигать другие потоки выше их базового приоритета, если они содержат блокировки ресурсов, которые требуют более высокие потоки приоритетов, что является другой возможностью, почему вы можете столкнуться с переключателями контекста.

Даже тогда, как вы говорите, Windows не является оперативной ОС, и в любом случае не гарантируется честность приоритетов потоков.

Answer 3

Чтобы атаковать эту проблему по-другому, вам нужно иметь так много развязанных асинхронных операций? Может быть полезно подумать: вертикально разбивать работу (асинхронно обрабатывать фрагменты данных numCores из конца в конец) вместо горизонтальной разбивки работы (как сейчас, с каждым фрагментом данных, обработанным на 15 этапах развязки); синхронно связывая некоторые из ваших 15 этапов, чтобы уменьшить общее количество до меньшего числа.

Накладные расходы на межпоточную связь всегда будут нетривиальными. Если некоторые из ваших 15 операций выполняют только небольшой кусок работы, коммутаторы контекста укусят вас.

Answer 4

Выполнение 15 асинхронных операций есть быть асинхронным? т. е. вы вынуждены работать этим путем ограничения какой-либо библиотеки или у вас есть выбор для синхронных вызовов?

Если у вас есть выбор, вам необходимо структурировать приложение, чтобы использование асинхронности контролировалось параметрами конфигурации. Разница между асинхронными операциями, возвращающимися в другом потоке или синхронными операциями, возвращающимися в одном потоке, должна быть прозрачной в коде. Таким образом, вы можете настроить его, не меняя структуру кода.

Фраза «смущающая параллель» описывает алгоритм, в котором большая часть выполняемой работы, очевидно, независима и поэтому может быть выполнена в любом порядке, что упрощает параллелизацию.

Но вы «объединяете 15 асинхронных операций через порты и приемники». Это можно охарактеризовать как «неудобно последовательное». Другими словами, одна и та же программа может быть логически записана в одном потоке. Но тогда вы потеряете любой параллелизм для работы с ЦП, выполняемой между асинхронными операциями (при условии наличия какого-либо значения).

Если вы напишете простой тест, чтобы вырезать какую-либо значительную работу с ЦП и просто измерить время переключения контекста, тогда угадайте, что вы будете измерять вариацию времени переключения контекста, поскольку вы обнаружено.

Единственная причина для запуска нескольких потоков - это то, что у вас есть значительная часть работы для процессоров, и поэтому вы хотели бы поделиться ею между несколькими процессорами. Если отдельные куски вычислений достаточно коротки, то переключение контекста будет значительным накладным расходами на любой ОС. Разбивая свои вычисления на 15 этапов, каждый очень короткий, вы, по сути, просите ОС сделать много ненужного переключения контекста.