Очередь потоков

← →
DevilDevil © (2012-07-23 16:57) [0]

Здравствуйте уважаемые мастера

Существует задача, которую необходимо распараллелить на несколько процессоров. Для этого планируется создать очередь потоков, количество потоков = количеству ядер.

В цикле (основной поток):
- собираем некоторую инфо
- ждём первый освободившийся поток
- кидаем ему инфо и адрес куда он должен записать результат

Ну и соответственно когда всё инфо считанно:
- ждём обработки всех потоков
- удаляем все потоки

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

← →
Медвежонок Пятачок © (2012-07-23 17:05) [1]

- ждём первый освободившийся поток
- кидаем ему инфо и адрес куда он должен записать результат

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

← →
Сергей М. © (2012-07-23 17:08) [2]

> как узнать количество ядер

GetProcessAffinityMask()

> как грамотнее организовать эту очередь потоков

У каждого потока уже есть своя индивидуальная очередь, и не одна.

Очередь заданий на асинхронное выполнение пользовательских процедур управляется вызовами QueueUserAPC.

Очередь пользовательских сообщений потоку и принадлежащим ему окнам управляется асинхронными вызовами PostThreadMessage() и PostMessage() соответственно.

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

← →
MsGuns © (2012-07-23 17:34) [3]

Если у Вас потоки чего-то там ждут (особенно "команд") из главного потока, то ИМХО, у Вас совершенно неверное представление об их предназначении

← →
DevilDevil © (2012-07-23 17:57) [4]

Допустим имеется файл.
Допустим если прочитать его в память, проанализировать - получается 1000 кусков данных.

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

Сейчас 1 главный поток обрабатывает информацию, получает кусок данных и перерабатывает этот кусок данных в алгоритме.

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

Слушаю предложения по реализации

> Сергей М. © (23.07.12 17:08) [2]
обязательно ли явно привязывать SetThreadAffinityMask, не лучше ли доверить распределение потоков по ядрам ОС ?

← →
Юрий Зотов © (2012-07-23 18:09) [5]

> DevilDevil © (23.07.12 17:57) [4]
> Предполагается сделать так, чтобы главный поток осуществлял
> чтение исходных кусков и перенаправлял их обработку на потоки.

То есть, главный поток создает не очередь потоков, а очередь кусков. А далее см. [1].

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

Очередь, конечно, должна быть синхронизирована.

← →
Юрий Зотов © (2012-07-23 18:19) [6]

> DevilDevil © (23.07.12 17:57) [4]

Чтобы было проще.

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

← →
DevilDevil © (2012-07-23 18:42) [7]

> Юрий Зотов © (23.07.12 18:09) [5]
> То есть, главный поток создает не очередь потоков, а очередь
> кусков. А далее см. [1].

Мне кажется это приведёт к дополнительному расходу памяти. Эффективнее думаю, сделать как описал в [0]:
В цикле (основной поток):
- собираем некоторую инфо, получаем кусок
- ждём первый освободившийся поток
- кидаем ему кусок и адрес куда он должен записать результат

Ну и соответственно когда всё инфо считанно:
- ждём обработки всех потоков
- удаляем все потоки

Исходный кусок занимает в среднем в 4 раза больше, чем инфо. А результирующий кусок в среднем в 6 раз меньше чем инфо. То есть в 24 раза )

Именно очередь потоков.
Но как:
- организовать очередь и синхронизировать её
- поток должен сигнализировать что он "освободился"
- запустить поток в работу после ожидания своей очереди

← →
Медвежонок Пятачок © (2012-07-23 18:48) [8]

- ждём первый освободившийся поток

Вот нахрена спрашивается?

← →
DevilDevil © (2012-07-23 18:53) [9]

> Медвежонок Пятачок © (23.07.12 18:48) [8]

перезадай вопрос корректно

← →
Anatoly Podgoretsky © (2012-07-23 19:10) [10]

GetProcessAffinityMask
http://msdn.microsoft.com/en-us/library/windows/desktop/ms683213(v=vs.85).aspx

← →
Сергей М. © (2012-07-23 20:37) [11]

> не лучше ли доверить распределение потоков по ядрам ОС ?

А зачем тогда вообще задумываться про "ядра" ?

← →
Rouse_ © (2012-07-23 21:18) [12]

Достаточно бестолковая задача, ось все равно сделает так, как ей более удобно. Нет, ну она конечно постарается придерживаться условиям, заданной при SetThreadAffinityMask, но не удивляйся моменту когда все нити поочередно будут исполнятся на одном ядре или прыгать с ядра на ядро. Года два назад была примерная тема, я там пример писал, который кажет сей нюанс.
Оть если твой код использует расширенные возможности процессора (ну там устаревший MMX или SSE) тут уже более реально (почему - линк искать лениво - в инете все есть), по поводу банального прикладного кода, плюнь и забудь, по любому ось быстрее и правильней сделает.

← →
Inovet © (2012-07-23 21:32) [13]

> [11] Сергей М. © (23.07.12 20:37)
> А зачем тогда вообще задумываться про "ядра" ?

А сколько потоков создавать 1-2 или 100, может 1000000.

← →
Rouse_ © (2012-07-23 21:46) [14]

> Inovet © (23.07.12 21:32) [13]
> А сколько потоков создавать 1-2 или 100, может 1000000.

Все зависит от конкретной задачи и возможностей конкретной машины.
Начиная с определенного количества нитей (потоков) производительность начинает падать.
Т.е. грубо при одной нити мы имеем 100 вычислений в секунду, при четырех 400 при восьми всего 600, значит золотая серединка где-то почти рядом :)

← →
Rouse_ © (2012-07-23 21:54) [15]

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

← →
Rouse_ © (2012-07-23 21:57) [16]

ЗЫ: алго естественно асинхронное, дабы не потерять производительность на ожидании окончания работы каждой нити на каждой итерации. :)

← →
DevilDevil © (2012-07-23 22:32) [17]

> Rouse_ © (23.07.12 21:57) [16]

мне тут на одном форуме сказали, что оптимально - рабочих потоков создавать столько сколько ядер. Потоки могут прыгать по ядрам, но если им явно задать SetThreadAffinityMask - не должны. Другое дело, хочется мне сделать претензию на кроссплатформ, поэтому завязываться на API ОС честно говоря не хочется. Я поэтому и спросил, насколько это оправдано

Но суть вопроса больше не в этом, а как именно по умному синхронизировать такую очередь.

← →
Rouse_ © (2012-07-23 22:57) [18]

> Но суть вопроса больше не в этом, а как именно по умному
> синхронизировать такую очередь.

По умному - не учитывать количество ядер и делать все по классике.

← →
Юрий Зотов © (2012-07-23 23:39) [19]

> DevilDevil © (23.07.12 22:32) [17]
> как ... синхронизировать ... очередь.

Дык... как любой список - критической секцией, например. Или взять готовый TThreadList.

← →
DevilDevil © (2012-07-24 00:41) [20]

> Rouse_ © (23.07.12 22:57) [18]
> По умному - не учитывать количество ядер и делать все по
> классике.

это наоборот по глупому. Для ресурсоёмких обсчётов не использовать всю мощность компьютера - глупо

> Юрий Зотов © (23.07.12 23:39) [19]
давайте опишу типичные ситуации. Скажите как реализовать
- нужно создавать потоки в "спящем" режиме. По идее suspended, но не знаю, насколько это корректно
- т.е. изначально количество свободных потоков = N. Необходимо иметь свойство "Количество свободных потоков"
- если в очереди свободных потоков, то дождаться первого свободного
- изъять первый свободный поток
- "запустить" поток ?
- когда поток отработал - он должен сам себя пометить как освободившийся и засунуть себя в очередь. приостановить ?
- дождаться пока число свободных потоков не равно N

Вот эти моменты я не знаю точно как правильно реализовать.
Не совсем понимаю как правильно использовать критическую секцию. Ей обязательно перекрывать функцию, или можно обойтись без неё, если используется простое обращение к полю:
property Count: integer read FCount;

Вообще можно ли в каких-то случаях обойтись без критической секции, синхронизировать как то boolean-ом ? Может ли произойти неправильное чтение из кэша. Например если 2 потока на разных ядрах обращаются к одному участку памяти. В первом ядре я пишу в глобальную переменную значение (и значение сидит в кэше этого ядра), а на другом ядре я читаю из этой ячейки (и читаю уже из кэша своего ядра).

Не совсем понимаю, зачем нужны другие синхронищирующие сущности кроме критической секции. И не понимаю как правильно организовать простои и ожидания для потоков. Через while (условие) do; Или через suspend. Или может через sleep. Или через WaitForSingleObject

В общем проясните пожалуйста эти азы. Или дайте толковую статейку

← →
Rouse_ © (2012-07-24 01:12) [21]

> это наоборот по глупому. Для ресурсоёмких обсчётов не использовать
> всю мощность компьютера - глупо

Боюсь ты не внимательно прочитал мой пост за номером [12]
Никто тебя не лишит возможности использовать железо на всю его дурь :)
Тебе просто могут помешать сделать это нерациональным способом, с точки зрения системы.
По поводу глупости - спасибо, буду работать над собой.

Теперь по пунктам:
КС необходимо использовать при записи/чтении критических для модификации блоков данных. Если ты читаешь данные размером до 4 байт, то, как правило, такие операции атомарны, т.е. синхронизация не нужна (с оговоркой - при чтении).

В частности в этом вот моменте:

> Вообще можно ли в каких-то случаях обойтись без критической
> секции, синхронизировать как то boolean-ом ? Может ли произойти
> неправильное чтение из кэша. Например если 2 потока на разных
> ядрах обращаются к одному участку памяти. В первом ядре
> я пишу в глобальную переменную значение (и значение сидит
> в кэше этого ядра), а на другом ядре я читаю из этой ячейки
> (и читаю уже из кэша своего ядра).

тут немного сумбурно.
любой блок данных до 4 байт (допустим в нем будет булевый параметр) может быть атомарно изменен при помощи Interlocked функций.

По поводу КЭШ-а не ясно что имеется ввиду, TLB? Ну тут вообще не наша епархия.
Локальный стэк нити? Если да, то нет - разные адреса.
Если TLS - то опять-же разные адреса.
Если Heap - то тут уже все зависит от реализации кода.

> В общем проясните пожалуйста эти азы. Или дайте толковую
> статейку

Тут я думаю нужно начать с Рихтера.
http://rouse.drkb.ru/books/rihter.zip

← →
Сергей М. © (2012-07-24 10:05) [22]

> нужно создавать потоки в "спящем" режиме

Ты имеешь ввиду suspended mode ?
Это лишь усложнит задачу.
Создавай в нормальном (running mode). Сразу после получения управления поток организует цикл, в теле которого он "вычерпывает" инф-цию из ассоциированной с ним APC-очереди или очереди сообщений, переходя в ожидании при пустой очереди в kernel mode, дабы не тратить бездарно кванты процессорного времени (SleepEx, MsgWaitForMultipleObjectsEx - для APC-очереди, Get/WaitMessage - для очереди сообщений)

> Все зависит от конкретной задачи и возможностей конкретной
> машины.
> Начиная с определенного количества нитей (потоков) производительность
> начинает падать.
> Т.е. грубо при одной нити мы имеем 100 вычислений в секунду,
> при четырех 400 при восьми всего 600, значит золотая серединка
> где-то почти рядом :)

Доверим это основному потоку. Если вычисления действительно длительные можно собирать статистику о количестве вычисленных итераций в единицу времени и самостоятельно регулировать число потоков :)

P.S. Где тут воробьи?

> Если вычисления действительно длительные можно собирать
> статистику о количестве вычисленных итераций в единицу времени
> и самостоятельно регулировать число потоков :)

Дык собственно так оно обычно и делается :)
Я в свое время эксперементировал: http://www.delphimaster.net/view/15-1178118027

← →
Андреевич (2012-07-24 21:07) [26]

> мне тут на одном форуме сказали, что оптимально - рабочих
> потоков создавать столько сколько ядер.

наделать по кол-ву ядер, но с привязкой не играться, самое простое

> Дык собственно так оно обычно и делается :)
>

Ну дык,

> это наоборот по глупому.

Вот и результат :) Надо было по количеству ядер делать и железо бы целее было ;)

> Надо было по количеству ядер делать и железо бы целее было ;)

Тю, маэстро - что есть железо, зато каков был эффект :)))

← →
han_malign (2012-07-26 09:59) [29]

> Я в свое время эксперементировал

- I/O Completion Ports...
- с указанием количества конкурентных потоков по числу ядер, и можно делать потоков сколько угодно(с разумным ограничением по памяти) - "лишние" разблокируются только если один из уже запущенных завершается или переходит в состояние ожидания(не связанное с целевым GetQueuedCompletionStatus)...

Посмотри OmniThreadLibrary, там многие полезные возможности реализованы

Очередь потоков Найти похожие ветки