Есть у меня такая, казалось бы, очень простая проблема: есть некий источник задач, каждую из которых можно решать параллельно. Первое что приходит на ум, так это ставить задачи в очередь, а потом в несколько потоков доставать их из неё и выполнять. Пожалуй, всем известно решение классической задачи про очередь (в которую могут как писать так и читать несколько потоков) на основе Хоаровского Монитора:

class SimpleHoareMonitorQueue
{
  //...

  public void Enqueue (object task)
  {
    lock (this)
    {
      ПоложитьЗаданиеВОчередь(task);
      System.Threading.Monitor.Pulse(this);
    }
  }

  public object Dequeue ()
  {
     lock (this)
     {
       while (очередь пуста) System.Threading.Monitor.Wait(this);
       return ДостатьЗаданиеИзОчереди();
     }
  }
}

Так вот, всё это конечно хорошо, но как быть если:
1) задачи могут ставиться в очередь по 15 миллионов штук в секунду,
или как быть если
2) за доступ к такой очереди конкурируют 500 потоков?

Проблема в том, что в main-stream операционках (Windows/Linux):
1) невыгодно посылать Pulse чаше чем несколько сотен раз в секунду,
и
2) невыгодно лочить более десятка потоков на один и тот же монитор...

Есть ещё один момент. Время выполнения каждой задачи заранее не известно. Возможно, что задача взятая из очереди окажется пустышкой (например, вызов виртуальной функции, которая "меняет пару полей") т.е. выполнится за 10 тактов или меньше, а возможно, что задача окажется очень долговыполняемой (установить соединение с базой данных, послать SQL запрос и дождаться ответа).

И как быть? Какую стратегию придумать, чтобы можно было оба пункта (1) и (2) победить одновременно?



Я придумал одно решение, но оно мне не очень нравится. Но, похоже что другого просто нет...

Решение такое (этакая "самоорганизующаяся система" )

А) Создаём список трудящихся потоков (самодельный пул потоков) каждый из которых делает вот что:

Лезет в очередь, достаёт из неё очередную задачу и выполняет её. Если очередь оказалась пустой, то поток залочивается на своём собственном System.Threading.AutoResetEvent и ждёт пока его не разлочит кто-то другой. Обращаю внимание, что он спит "не на очереди", а сам.

Б) Создаём контроллирующий высокоприоритетный поток, который в цикле делает вот что:

Тупо спит 10 миллисекунд, потом просыпается и если очередь задач не пуста, то пытается разбудить один поток из списка трудящихся потоков. Если оказывается, что все потоки уже разбужены и трудятся, то создаёт ещё несколько потоков, добавляя их в список (если конечно не превышен максимум количества потоков).

Модификация:
Поток залоченный на своём System.Threading.AutoResetEvent не до бесконечности ждёт пока его разлочит кто-то другой, а секунд эдак 30, а потом думая, что раз в его услугах больше никто не нуждается, умирает. Контролирующий поток удаляет умершие потоки из списка.


Короче получается вот что:

1) Если задачи выполняются очень быстро, то вся эта система динамически скатывается в состояние в котором в списке трудящихся потоков их всего две штуки — один постоянно активен, а второй будится каждые 10 миллисекунд контроллирующим потоком, но быстро вырубается, так как ему не хватает задач в очереди -- всё успевает сделать первый поток. На производительность эти 10 миллисекундные "тычки" практически не влияют.

2) Если задачи выполняются долго, то вся эта система динамически скатывается в противоположное состояние -- список трудящихся потоков имеет максимальный размер и все потоки активно трудятся.

Время перехода системы из состояния (1) в состояние (2) ограничено тем, сколько раз в секунду контролирующий поток пробуждает/создаёт новые потоки. Если 100 раз в секунду, то до активации 500 потоков пройдёт 5 секунд. Время перехода из состояния (2) в состояние (1) практически мгновенное — как только задачи в очереди кончились, так все трудящиеся потоки и залочились (а спустя 30 секунд совсем умерли).

Я написал тестовую програмку. Это работает. Действительно на медленных задачах трудятся 500 потоков, а на быстрых 1 поток по 15-16 миллионов задач в секунду.

Есть ли какие-то другие эвристики позволяющие решить эту задачу более оптимально?

Например, как контролирующему потоку догадаться, что надо разбудить не 1 поток, а, например, 100 если навалилась куча работы? То есть как уменьшить время перехода этой саморегулирующейся системы из состояния (1) в состояние (2) в случае пиковой нагрузки?

Здравствуйте, С. Ю. Губанов, Вы писали:

для начала нужно понять, что если у вас 1000 тредов равного приоритета и частота переключений тредов — 1 кгц, в реальности каждый тред будет получать управление 1 раз в секунду на 1 ms.
число физ ядер отличное от 1 задачу сильно не упрощают, поскольку их все равно много меньше чем число тредов, и правило сохраняется.
в таких условиях треды начинают взаимодействовать очень натужно.
например тред А посылает треду В сообщение, и В сразу вам посылает ответ,.. ответ вы получите через две секунды.
откуда возникает в системе 15 миллионов задач в секунду?
см. замечание выше. если все сотни тредов крутятся на одном проце, вся система будет сильно тормозить, и 15 миллионов одна из ее частей не обеспечит.
если ж эти 15 миллионов идут со внешних устройств, то нужно применять буфер-накопитель — если конкретно очередь или циклический буфер, возможно применять двухбуферную схему. но это тоже зависит от специфики вашей системы.

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

Здравствуйте, merk, Вы писали:

M>Здравствуйте, С. Ю. Губанов, Вы писали:

M>для начала нужно понять, что если у вас 1000 тредов равного приоритета и частота переключений тредов — 1 кгц, в реальности каждый тред будет получать управление 1 раз в секунду на 1 ms.
M>число физ ядер отличное от 1 задачу сильно не упрощают, поскольку их все равно много меньше чем число тредов, и правило сохраняется.
M>в таких условиях треды начинают взаимодействовать очень натужно.
M>например тред А посылает треду В сообщение, и В сразу вам посылает ответ,.. ответ вы получите через две секунды.
M>откуда возникает в системе 15 миллионов задач в секунду?
M>см. замечание выше. если все сотни тредов крутятся на одном проце, вся система будет сильно тормозить, и 15 миллионов одна из ее частей не обеспечит.
M>если ж эти 15 миллионов идут со внешних устройств, то нужно применять буфер-накопитель — если конкретно очередь или циклический буфер, возможно применять двухбуферную схему. но это тоже зависит от специфики вашей системы.

M>также нужно заметить что в задачах жесткого рилтайма загрузка проца не должна быть под 100 процентов, иначе у вас будет падать время реакции. если вы надеетесь написать систему, что будет быстрорегарирующий, и с загрузкой под 100 процентов — это нереально.

тьфу...посмотрел ваш пост по диагонали. вы и сами отметили факт — "раз в секунду". прошу пардону.

Здравствуйте, С. Ю. Губанов, Вы писали:

поскольку для сглаживания нагрузки накопитель нужно использовать по любому, то 15 миллионов задач нужно запихивать в накопитель самым простым образом
пусть тред А поставляет задачи в буфер. а буфер — есть двухсвязный список задач, например.
тогда тред А просто делает из текущих задач собственный список, без всяких блокировок, поскольку владеет им монопольно. потом, по таймеру или еще какому условию, лочит общий накопитель, и вставляет список туда, физически просто подлинковывая хвост и голову в список накопителя. это эквмвалентно вставке двух элементов.
никакой супер скорости тут не нужно, делается без приколов с интерлоками, на простых мьютексах или еще чем
вопрос с заталкиванием 15 миллионов надеюсь закрыт.
остается вопрос с вытаскиванием 15 миллионов. их тоже можно вытаскивать кусками-списками. а сам накопитель будет несколько более сложным "списком списков".

Здравствуйте, merk, Вы писали:

M>для начала нужно понять

Так 15 миллионов/сек, это один из предельных случаев, когда выполнение задачи сводится к постановке другой задачи в очередь, и так далее по кругу. Такая скорость получается только при одном работающем потоке.

Когда задачи выполняются долго -- это другой предельный случай, и скорость конечно значительно меньше (сотни задач в секунду).

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

Практически, у меня больше половины задач будут выполняются почти "за пару тактов", но процентов может быть десять или двадцать будут выполняются долго -- полезут в базу данных (кстати, пока они ожидают ответа от базы данных, они заблокированы; ответ от MySQL обычно приходит не раньше чем через 2-3 миллисекунды, т.е. как минимум по 2-3 миллисекунды после каждого SQL запроса поток испольняющий "долгую" задачу будет тупо спать).

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>Я написал тестовую програмку. Это работает. Действительно на медленных задачах трудятся 500 потоков, а на быстрых 1 поток по 15-16 миллионов задач в секунду.

Что понимается в данном случае под задачей (которые успевают выполняться по 15M штук в секунду)?
И на какой аппаратной/программной платформе был получен данный результат?

Здравствуйте, merk, Вы писали:

M>пусть тред А поставляет задачи в буфер.

Да, есть конечно треды поставщики задач приходящих из внешнего Мира,
но ещё есть задачи которые генерятся внутри системы во время решения предыдущих задач,
ещё есть задачи генерируемые "по таймеру" и т.д.

Здравствуйте, eao197, Вы писали:

E>Что понимается в данном случае под задачей (которые успевают выполняться по 15M штук в секунду)?
E>И на какой аппаратной/программной платформе был получен данный результат?

AMD Athlon64 X2 2.21 GHz, Windows XP SP3.
Intel Xeon (2 ядра * 2 гипертреда) 2.8 GHz, Linix/Mono -- на нём работает раза в два помедленее -- рухлядь выбрасывать пора.

Для пустой циркуляции сообщений по кругу уже докрутил до пиковых 33 миллионов в секунду на Атлоне (правда не стабильно , видимо 15-16 -- это когда фактически работает только один проц, но иногда параллельно работают два проца, вот и получается 33М, вот ломаю голову как бы это постабильнее на двух процах заработало, пока не понял как):

namespace MFISoft
{
    internal static class Test1
    {
        private sealed class Node: Runtime.Object
        {
            public Node next;

            public Node (Node next)
            {
                this.next = next;
            }

            protected override void ProcessMessage (object message, Runtime.Object sender)
            {
                this.next.EnqueueMessage(message, this);
            }

            public void Start ()
            {
                this.next.EnqueueMessage(new object(), this);
            }
        }

        private static Node NewRing (int n)
        {
            Node last = new Node(null);
            Node first = last;
            for (int i = 0; i < n; i++)
            {
                first = new Node(first);
            }
            last.next = first;
            return first;
        }

        private static Node ring;

        public static void Do ()
        {
            ring = NewRing(12);
            for (int i = 0; i < 12; i++)
            {
                ring.Start();
                System.Threading.Thread.Sleep(1);
            }
            System.Console.ReadLine();
        }
    }
}

Я сейчас сочиняю систему объектов обменивающихся сообщениями асинхронно. У каждого MFISoft.Runtime.Object есть очередь входящих сообщений. Чтобы отправить сообщение, нужно вызвать recipient.EnqueueMessage(message, sender), сообщение будет добавлено в очередь входящих сообщений recipient-а, а сам recipient (если этого ещё не было сделано ранее) будет добавлен в очередь объектов ожидающих процессорного времени — это и есть та самая очередь задач о которой я писал. Когда recipient дождётся процессорного времени, то для каждого сообщения находящегося у него в очереди (на момент начала обработки) будет вызван метод ProcessMessage(message, sender). В данном тесте, в обработчике сообщения, сообщение тупо пересылается следующему получателю по кругу.
Система гарантирует, что сообщения отправленные получателю будут доставлены до него именно в том порядке в каком были посланы, а так же что обработчик ProcessMessage одновременно будет вызван только из одного потока, т.е. на прикладном уровне никаких синхронизаций делать уже не надо. Для разных объектов сообщения доставляются асинхронно.

Тот же самый тест с другими параметрами:

namespace MFISoft
{
    internal static class Test1
    {
        private sealed class Node: Runtime.Object
        {
            ...

            protected override void ProcessMessage (object message, Runtime.Object sender)
            {
                System.Threading.Thread.Sleep(20);
                this.next.EnqueueMessage(message, this);
            }

            ...
        }

        ...

        public static void Do ()
        {
            ring = NewRing(100000);
            for (int i = 0; i < 1000; i++)
            {
                ring.Start();
                System.Threading.Thread.Sleep(1);
            }
            System.Console.ReadLine();
        }
    }
}

проверяет систему в другом режиме: здесь 1000 потоков осуществляют по 47'000-48'000 событий пересылки сообщений в секунду.

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>Я сейчас сочиняю систему объектов обменивающихся сообщениями асинхронно. У каждого MFISoft.Runtime.Object есть очередь входящих сообщений. Чтобы отправить сообщение, нужно вызвать recipient.EnqueueMessage(message, sender), сообщение будет добавлено в очередь входящих сообщений recipient-а, а сам recipient (если этого ещё не было сделано ранее) будет добавлен в очередь объектов ожидающих процессорного времени — это и есть та самая очередь задач о которой я писал. Когда recipient дождётся процессорного времени, то для каждого сообщения находящегося у него в очереди (на момент начала обработки) будет вызван метод ProcessMessage(message, sender). В данном тесте, в обработчике сообщения, сообщение тупо пересылается следующему получателю по кругу.
СЮГ>Система гарантирует, что сообщения отправленные получателю будут доставлены до него именно в том порядке в каком были посланы, а так же что обработчик ProcessMessage одновременно будет вызван только из одного потока, т.е. на прикладном уровне никаких синхронизаций делать уже не надо. Для разных объектов сообщения доставляются асинхронно.

насколько я понял, у вас получатель добавляется в очередь ожидающих "процессорного времени"... а когда он стартанет, из какого кода? есть какой-то тред, что в своем контексте крутит этих — получателей? если так — будет проблема.
также, а когда получатель выйдет их списка "ожидающих процесорного...".
система напоминает список EventListener'ов что-ли?

Здравствуйте, С. Ю. Губанов, Вы писали:
короче если у вас системы аля хуки(это ваши "задачи") передаваемые некоторым тредам для исполнения, то возникнет суровый вопрос блокировки.
например если тред А должен выполнить задачи В,С но эти задачи зависимы. если бы они исполнялись разными тредами — они бы исполнились, но если одним — нет. ну вы понимаете.
то есть крутить несколькими тредами или одним, через такие вот поручаемые им хуки можно только полностью независимые задачи.

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>

СЮГ>        public static void Do ()
СЮГ>        {
СЮГ>            ring = NewRing(12);
СЮГ>            for (int i = 0; i < 12; i++)
СЮГ>            {
СЮГ>                ring.Start();
СЮГ>                System.Threading.Thread.Sleep(1);
СЮГ>            }
СЮГ>            System.Console.ReadLine();
СЮГ>        }
СЮГ>    }
СЮГ>}
СЮГ>

Не очень понятно, как вы оцениваете время и объем проделанной работы. Вы создали кольцо из 12 узлов и запустили в нем 12 циклов (бесконечных) передачи сообщений. Как же подсчитывается количество переданных сообщений и затраченное на них время?

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>Есть у меня такая, казалось бы, очень простая проблема: есть некий источник задач, каждую из которых можно решать параллельно.

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

СЮГ>Есть ещё один момент. Время выполнения каждой задачи заранее не известно. Возможно, что задача взятая из очереди окажется пустышкой (например, вызов виртуальной функции, которая "меняет пару полей") т.е. выполнится за 10 тактов или меньше, а возможно, что задача окажется очень долговыполняемой (установить соединение с базой данных, послать SQL запрос и дождаться ответа).

СЮГ>Б) Создаём контроллирующий высокоприоритетный поток, который в цикле делает вот что:
СЮГ>Тупо спит 10 миллисекунд, потом просыпается и если очередь задач не пуста, то пытается разбудить один поток из списка трудящихся потоков. Если оказывается, что все потоки уже разбужены и трудятся, то создаёт ещё несколько потоков, добавляя их в список (если конечно не превышен максимум количества потоков).

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

СЮГ>Модификация:
СЮГ>Поток залоченный на своём System.Threading.AutoResetEvent не до бесконечности ждёт пока его разлочит кто-то другой, а секунд эдак 30, а потом думая, что раз в его услугах больше никто не нуждается, умирает. Контролирующий поток удаляет умершие потоки из списка.

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

СЮГ>Короче получается вот что:

СЮГ>1) Если задачи выполняются очень быстро, то вся эта система динамически скатывается в состояние в котором в списке трудящихся потоков их всего две штуки — один постоянно активен, а второй будится каждые 10 миллисекунд контроллирующим потоком, но быстро вырубается, так как ему не хватает задач в очереди -- всё успевает сделать первый поток. На производительность эти 10 миллисекундные "тычки" практически не влияют.

СЮГ>2) Если задачи выполняются долго, то вся эта система динамически скатывается в противоположное состояние -- список трудящихся потоков имеет максимальный размер и все потоки активно трудятся.

СЮГ>Время перехода системы из состояния (1) в состояние (2) ограничено тем, сколько раз в секунду контролирующий поток пробуждает/создаёт новые потоки. Если 100 раз в секунду, то до активации 500 потоков пройдёт 5 секунд. Время перехода из состояния (2) в состояние (1) практически мгновенное — как только задачи в очереди кончились, так все трудящиеся потоки и залочились (а спустя 30 секунд совсем умерли).

задача быстрой активации решается классически экспоненциальным активированием. первый раз активируются например 2 треда, второй(если предыдущий раз тоже была активация) — 4, и так далее. ну в разумных пределах разумеется. не стоит доходить до высоких степеней.

СЮГ>Я написал тестовую програмку. Это работает. Действительно на медленных задачах трудятся 500 потоков, а на быстрых 1 поток по 15-16 миллионов задач в секунду.

СЮГ>Есть ли какие-то другие эвристики позволяющие решить эту задачу более оптимально?

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

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>Есть у меня такая, казалось бы, очень простая проблема: есть некий источник задач, каждую из которых можно решать параллельно. Первое что приходит на ум, так это ставить задачи в очередь, а потом в несколько потоков доставать их из неё и выполнять. Пожалуй, всем известно решение классической задачи про очередь (в которую могут как писать так и читать несколько потоков) на основе Хоаровского Монитора:


Возможно будет интересно CDS:
http://gzip.rsdn.ru/forum/message/2993954.1.aspx

Автор: remark
Дата: 20.06.08

Там есть многопоточная FIFO очередь, спин-ожидание и др.

Здравствуйте, С. Ю. Губанов, Вы писали:

C#? Подожди, а как же я? Я ведь тоже это могу. Активные объекты. Забыл, да? Эх ты!

Здравствуйте, eao197, Вы писали:

E>Не очень понятно, как вы оцениваете время и объем проделанной работы. Вы создали кольцо из 12 узлов и запустили в нем 12 циклов (бесконечных) передачи сообщений. Как же подсчитывается количество переданных сообщений и затраченное на них время?

Я показал только код пользовательского уровня.
Подсчёт ведётся внутри Runtime.

СЮГ> Я придумал одно решение, но оно мне не очень нравится. Но, похоже что другого просто нет...

Поскольку очередь задач была одна, то одновременно нельзя было отправить более одного сообщения каким бы малым не было время его отправления. Разбил очередь задач на несколько параллельный очередей (96 штук). Каждый recipient привязан только к одной из них (связывается в конструкторе раз и навсегда). Теперь одновременно с большой вероятностью можно отправить несколько сообщений -- столько сколько процессоров в машине (несколько сообщений можно отправить получателям привязанным к разным очередям, чем больше очередей, тем больше вероятность "параллельности"). Да появились дополнительные накладные расходы, но они не очень большие, зато производительность теперь растет линейно с количеством процессоров. В итоге имею 18-27 миллионов сообщений в секунду на Athlon 64 X2 2.21 GHz. На четырёхядерном значит ожидаю, что-то под 40-50 Мегасообщений/сек, но у меня его нет, проверить не могу.

Здравствуйте, С. Ю. Губанов, Вы писали:

СЮГ>> На четырёхядерном значит ожидаю, что-то под 40-50 Мегасообщений/сек, но у меня его нет, проверить не могу.
Супер! Даже Д.Ю.Пучков не успевает отправлять более 5-7 мегасообщений в день!