Доброго времени суток !
Есть потребность передавать данные от одного потока другому наиболее быстрым способом.
Идея такая:
Поток, который отдаёт данные (пусть будет DataProducer)

• выделяет память в "куче" и получает указатель на эту память
• локирует буфер
• запихивает указатель в массив
• разлокирует буфер

Поток, который принимает данные (пусть будет DataReceiver)

• локирует массив указателей
• копирует их в свой массив
• разлокируется
• складывает последовательно все данные в один выходной буфер

Код написан (приведён ниже), идея работает.
Вопрос: можно ли как-нибудь ускорить этот процесс ?
Спасибо.



Это сообщение отредактировал(а) borisbn - 15.5.2011, 13:59

а что, медленно работает? вроде как нет никаких очевидных тормозов в коде..

не то чтобы медленно, но поток входных данных очень большой (~4 МБ/сек), плюс к тому таких потоков 48. Остальные места проекта я как-то "вылизал", а за это (немаловажное) переживаю... может есть вариант побыстрее ?

Пробовал перейти к raw-указателям (без STL) - получилось ещё медленнее 

Хм. Зачем лочить указатели? Выделение памяти (через new) давно уже должно быть непроблемным в многопоточной проге (т.е. два одновременно обратившихся за выделением памяти потока не должны получить указатель на одну и ту же область памяти).

Да и разговор о наиболее быстром способе при использовании библиотеки классов не имеет смысла. Создание и последующее неявное автоматическое удаление локальных переменных-экземпляров классов (в Вашем случае имеется в виду строка QMutexLocker locker( &m_mutex ), но при росте размера синхронизирующего кода подобных строк может стать больше) может приводить к большим накладным расходам из-за "тяжелых" конструктора и/или деструктора. Т.е. надо лезть и читать исходники библиотеки классов, чтобы гарантировать отсутствие тормозов по этой причине.
А вот явное использование функций синхронизации потоков и функций межпоточного обмена данными из WinAPI или аналогичных юниксовых примитивов будет гарантированно быстрее (если, конечно, выбрать адекватные задаче низкоуровневые функции и правильно всё реализовать). Нужна многоплатформенная переносимость - так #ifdef'ы и подобные директивы препроцессора никто не отменял.

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

да. иначе можешь получить среднее состояние.


так. минутку..
в потоке производителе ты что-то выполняешь. потом и него, передаешь указатель в поток потребитель. я правильно понял? если да - то тут нечего оптимизировать.

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

Добавлено через 3 минуты

указатели и не лочаться ;)
лочиться участок кода, добавляющий/удаляющий их из/в STL контейнер.

Я, наверное, неправильно выразился. Я лочу не указатели, а вектор, который их содержит. Вернее операции с этим вектором (push_back в одном потоке, begin, end, clear - в другом).


Посмотрел QMutexLocker. Как и ожидал - простой до невозможности:

а вот QMutex - мдааааа. Пожалуй, и правда, воспользуюсь советом и перейду на низкоуровневые объекты ОС. Спасибо.


Не проблема. Итак,
имеется 48 устройств, записывающих данные в память PC по DMA (устройств вообще-то 6 по 8 каналов в каждом, но для простоты объяснения пусть будет 48 устройств).
Данные пишутся по кольцу. На каждое устройство создан один поток (я его назвал DataProducer), который получает уведомления о том, что очередной блок данных записан (не всё кольцо, а, допустим, 1/10-я). Уведомления поступают с частотой ~10 мс. Эти данные необходимо обработать, но обработка может занять 1 мс, а может 100, поэтому обрабатывать в этом же потоке нельзя, иначе он может потерять уведомление и данные "перетрутся" новыми. Данные копируются в SpeedBuffer (ф-ция add) и обрабатывающему потоку сообщается, что данные можно забирать.
Для обработки создано ещё 48 потоков (я их назвал DataReceiver), которые ожидают события от своего DataProducer'а, забирают данные (ф-ция take у SpeedBuffer'а) и обрабатывают.

Добавлено через 1 минуту и 27 секунд
boostcoder, походу лочить надо форум, когда набираешь текст   

Добавлено через 12 минут и 12 секунд

вот не уверен. если ты про

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

то это вообще выполняется в том же потоке, где и empty()

его все так называют 


каким образом это происходит?

Добавлено через 3 минуты и 30 секунд

ну как знаешь..

по прерыванию от драйвера взводится событие. Он (DataProducer) "висит" на этом собитии.

это-то меня и интересовало. каким образом он висит на нем?

WaitForSingleObject

оно одно для всех 48ми каналов? или на каждый канал свое?

Jedem das Seine 
48 устройств - 48 WaitForSingleObject'ов

ну тогда все можно сильно упростить 
чуть позже опишу..

на самом деле всё немного по-другому: устройств 6, и Wait'ов тоже шесть. У каждого устройства 8 каналов, которые записываются в один массив. При получении уведомления каждый из шести потоков делает демультиплексирование этих данных на 8 каналов, и последовательно вызывает add у 8-ми обработчиков.

Можно слегка ускорить вычитывание данных если сделать накопление данных не в виде массива векторов, а в виде одного массива (как сделано при чтении). А потом при чтении просто делать swap массивов - аккумулятора и выходного массива (выходной перед этим понулить - позвать clear() )

Я бы всё-таки Продюсера сделал одним потоком - дал ему немного повыше приоритет, и пусть он в вечном цикле крутит цикл проверки поступления новых данных от всех устройств. Проверил все устойства - sleep(0) и обратно к началу вечного цикла. Думаю, уведомления от устройства у Вас идут в виде изменения значения указателя в кольцевом буфере - поэтому разные скорости поступления данных от разных устройств одним потоком обработать будет можно (поток на каждое устройство сохраняет значение проверенной позиции в кольцевом буфере, на следующей итерации сравнивает старое значение с текущим, и если есть расхождения - копирует кусок свежих данных в буфер соответствующего обрабатывающего потока).

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

Ну и всю эту часть (синхронизация потоков и передача данных между ними) делал бы "вручную" без всяких лишних библиотек классов. При этом будет возможен ещё один вариант ускорения - если буферы данных выровнять на границы параграфов, то можно будет копировать по 8 или по 16 байт одной командой (с использованием ММХ или SSE). Но надо будет или функцию копирования написать свою (вместо стандартной memcpy или виндовозовской CopyMemory), или взять оптимизированную под современные процессоры функцию копирования данных (у Агнера Фога была библиотечка на такую тему), или прописать копирование данных на чистом С (а там уж пусть компилятор старается-оптимизирует этот участок кода).
Т.е. если будет идти большая нагрузка на проц при "отключенной" реальной обработке данных Ресиверами - то может быть копирование данных от Продюсера к Ресиверу тормозит (библиотеки классов - такие библиотеки...). 

не совсем. я получаю прерывание от устройства каждую 1/10-ю от длины буфера. поток Продюсера "висит" на событии от драйвера (WaitForSingleObject - в user-space, SetEvent - в kernel), и делать бесконечный цикл (кушающий процессор) мне нет необходимости, а sleep(0) в Windows (проверено!) может кушать практически сколько угодно времени :(


Эти "качели" у меня получаются "нахаляву" из-за того, что драйвер выдаёт сообщение о заполнении буфера, причём у меня получается не два буфера, как Вы предлагаете, а 10.

И ещё: IMHO в моём варианте лоченье происходит на меньшее время, т.к. лочится только код копирующий указатели, а не массивы

Добавлено через 2 минуты и 51 секунду

Тогда мне придётся лочить этот входной массив на время копирования в него, а сейчас у меня лочится только копирование указателей...


аууууу 

Если вам очень критична скорость, то имеет смысл вообще отказаться от stl контейнеров (в этом месте)
Нужно сделать контейнер в котором можно разделить процессы выделения места (на это время контейнер будет лочится) и перенесение информации в это выделенное место (на это время контейнер лочить не надо)

Это должна быть несколько модифицированная queue, с методами:

•  Алоцировать место в хвосте (alloc_push) - выделяет место для записи данных
•  Записать данные (push_write) - копирует данные в выделенное в п1 место
•  Опубликовать (push) - делает буфер и п1 доступным для чтения
•  Зарезервировать место для чтения (alloc_pop) - выдает размер записанных данных
•  Читать (pop_read) - читает данные из области, возвращенной методом alloc_pop
•  Пометить как прочтенные (pop) - возвращает буфер (из alloc_pop) для дальнейшей перезаписи

При такой организации вообще не надо копировать данные между очередями.

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

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

Резюмирую:
1. По объективным причинам (архитектура железа) данные от продюсера необходимо копировать в некий буфер, при чём делать это нужно без локирования
2. Локировать желательно функциями/объектами WinAPI, а не объектами Qt
3. Чтобы избавиться от копирования на принимающей стороне ( m_outputBuffer ), возможно лучше пользоваться примерно такой ф-цией

и переделать m_outPtrs на std::queue или на std::list

Не совсем так - у вас сейчас постоянно создаются и удаляются вектора для собственно данных. Я предлагал данные копировать непосредственно в очередь:

Чтение аналогично

Если сделать операции *push/*pop атомарными, то даже mutex не потребуется