Модераторы: Daevaorn

Поиск:

Ответ в темуСоздание новой темы Создание опроса
> Быстрый способ передать данные между потоками 
:(
    Опции темы
VictorTsaregorodtsev
Дата 17.5.2011, 00:14 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 274
Регистрация: 28.7.2006

Репутация: 1
Всего: 8



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

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

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

Ну и всю эту часть (синхронизация потоков и передача данных между ними) делал бы "вручную" без всяких лишних библиотек классов. При этом будет возможен ещё один вариант ускорения - если буферы данных выровнять на границы параграфов, то можно будет копировать по 8 или по 16 байт одной командой (с использованием ММХ или SSE). Но надо будет или функцию копирования написать свою (вместо стандартной memcpy или виндовозовской CopyMemory), или взять оптимизированную под современные процессоры функцию копирования данных (у Агнера Фога была библиотечка на такую тему), или прописать копирование данных на чистом С (а там уж пусть компилятор старается-оптимизирует этот участок кода).
Т.е. если будет идти большая нагрузка на проц при "отключенной" реальной обработке данных Ресиверами - то может быть копирование данных от Продюсера к Ресиверу тормозит (библиотеки классов - такие библиотеки...). 
PM MAIL WWW   Вверх
borisbn
Дата 17.5.2011, 09:12 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Эксперт
****


Профиль
Группа: Завсегдатай
Сообщений: 4875
Регистрация: 6.2.2010
Где: Ростов-на-Дону

Репутация: 22
Всего: 135



Цитата(VictorTsaregorodtsev @  17.5.2011,  00:14 Найти цитируемый пост)
Думаю, уведомления от устройства у Вас идут в виде изменения значения указателя в кольцевом буфере

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

Цитата(VictorTsaregorodtsev @  17.5.2011,  00:14 Найти цитируемый пост)
Обрабатывающему потоку же дал 2 буфера - когда заполнился один - поток начинает его обрабатывать

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

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

Добавлено через 2 минуты и 51 секунду
Цитата(xvr @  16.5.2011,  13:58 Найти цитируемый пост)
Можно слегка ускорить вычитывание данных если сделать накопление данных не в виде массива векторов, а в виде одного массива (как сделано при чтении). А потом при чтении просто делать swap массивов - аккумулятора и выходного массива (выходной перед этим понулить - позвать clear() )

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

Цитата(boostcoder @  15.5.2011,  16:03 Найти цитируемый пост)
ну тогда все можно сильно упростить чуть позже опишу..

аууууу smile


--------------------
Женщины отличаются от программистов тем, что у них чары состоят из стрингов
PM MAIL Jabber   Вверх
xvr
Дата 17.5.2011, 11:37 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Эксперт
****


Профиль
Группа: Комодератор
Сообщений: 7046
Регистрация: 28.8.2007
Где: Дублин, Ирландия

Репутация: 60
Всего: 223



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

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

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

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

PM MAIL   Вверх
borisbn
Дата 17.5.2011, 22:44 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Эксперт
****


Профиль
Группа: Завсегдатай
Сообщений: 4875
Регистрация: 6.2.2010
Где: Ростов-на-Дону

Репутация: 22
Всего: 135



Цитата

void SpeedBuffer< T >::add( const T * p, unsigned int count )
{
// Алоцировать место в хвосте (alloc_push) - выделяет место для записи данных
// Записать данные (push_write) - копирует данные в выделенное в п1 место
    Vect * v = new Vect( p, p + count ); 

    QMutexLocker locker( &m_mutex );
    m_inPtrs.push_back( v ); 
    m_size += count;

// Опубликовать (push) - делает буфер и п1 доступным для чтения (~QMutexLocker)
} 

template< class T >
const T * SpeedBuffer< T >::take( unsigned int & count )
{
    m_mutex.lock();
// Зарезервировать место для чтения (alloc_pop) - выдает размер записанных данных
    m_outPtrs.assign( m_inPtrs.begin(), m_inPtrs.end() );
    m_inPtrs.clear();
    count = m_size;
    m_size = 0;
    m_mutex.unlock();
    m_outputBuffer.clear();
    for ( typename VectVect::const_iterator it = m_outPtrs.begin(); it != m_outPtrs.end(); ++it )
    {
        const Vect * v = (*it);
// Читать (pop_read) - читает данные из области, возвращенной методом alloc_pop
        m_outputBuffer.insert( m_outputBuffer.end(), v->begin(), v->end() );
Пометить как прочтенные (pop) - возвращает буфер (из alloc_pop) для дальнейшей перезаписи
        delete v;
    }
    return &m_outputBuffer[ 0 ];
}


Цитата(xvr @  17.5.2011,  11:37 Найти цитируемый пост)
Следующей точкой оптимизации может стать отказ от копирования памяти вообще - сделать в драйвере пул памяти и отмэпировать его в пользовательское пространство. Драйвер будет общаться с клиентом пересылая ему указатели, смотрящие в этот пул

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

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

template< class T >
const std::vector< T > * SpeedBuffer< T >::takeOne()
{
    QMutexLocker( &m_mutex );
    const std::vector< T > * v = m_outPtrs.front(); // [offtopic] эээх, как auto не хватает :( [/offtopic]
    m_outPtrs.pop_front();
    m_size -= v->size();
    return v;
}

// использование
if ( buff.empty() == false ) {
    const std::vector< int > * v = buff.takeOne();
    обработать( &(*v)[ 0 ], v->size() );
    delete v;
}

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


--------------------
Женщины отличаются от программистов тем, что у них чары состоят из стрингов
PM MAIL Jabber   Вверх
xvr
Дата 18.5.2011, 11:50 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Эксперт
****


Профиль
Группа: Комодератор
Сообщений: 7046
Регистрация: 28.8.2007
Где: Дублин, Ирландия

Репутация: 60
Всего: 223



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

Код

template<class Item>
class SQueue {
public:
 class interator { ... };
 typedef pair<iterator,iterator> iterator_pair;

 iterator_pair alloc_push(size_t size);
 void push(iterator_pair);

 iterator_pair alloc_pop();
 void pop(iterator_pair); 
};


void SpeedBuffer< T >::add( const T * p, unsigned int count )
{
// Алоцировать место в хвосте (alloc_push) - выделяет место для записи данных
   SQueue<T>::iterator_pair i2; 
   {
      QMutexLocker locker( &m_mutex );
      i2 = squeue.alloc_push(count);
   }
// Записать данные (push_write) - копирует данные в выделенное в п1 место
  std::copy(p,p+count,i2.first);
// Опубликовать (push) - делает буфер и п1 доступным для чтения
   {
      QMutexLocker locker( &m_mutex );
      squeue.push(i2);
   }
} 
Чтение аналогично

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

PM MAIL   Вверх
Ответ в темуСоздание новой темы Создание опроса
Правила форума "С++:Общие вопросы"
Earnest Daevaorn

Добро пожаловать!

  • Черновик стандарта C++ (за октябрь 2005) можно скачать с этого сайта. Прямая ссылка на файл черновика(4.4мб).
  • Черновик стандарта C (за сентябрь 2005) можно скачать с этого сайта. Прямая ссылка на файл черновика (3.4мб).
  • Прежде чем задать вопрос, прочтите это и/или это!
  • Здесь хранится весь мировой запас ссылок на документы, связанные с C++ :)
  • Не брезгуйте пользоваться тегами [code=cpp][/code].
  • Пожалуйста, не просите написать за вас программы в этом разделе - для этого существует "Центр Помощи".
  • C++ FAQ

Если Вам понравилась атмосфера форума, заходите к нам чаще! С уважением, Earnest Daevaorn

 
0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема | C/C++: Общие вопросы | Следующая тема »


 




[ Время генерации скрипта: 0.0469 ]   [ Использовано запросов: 22 ]   [ GZIP включён ]


Реклама на сайте     Информационное спонсорство

 
По вопросам размещения рекламы пишите на vladimir(sobaka)vingrad.ru
Отказ от ответственности     Powered by Invision Power Board(R) 1.3 © 2003  IPS, Inc.