Russian Qt Forum

Здравствуйте!

Работаю на Qt4.8.6

У меня есть два параллельных потока.

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

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

Я думаю использовать вариант побыстрее. Сделать промежуточный буфер между этими потоками - QQueue. Один буфер общий на два потока. То есть один поток добавляет данные к очереди, а второй извлекает.

Подскажите, пожалуйста, какие могут быть подводные камни в таком решении. В доке написано, что QQueue - reentrant. То есть очередь защищена от проблем при использовании многопоточности. Но все, же, что еще нужно учитывать? Нужно ли делать синхронизацию между потоками?

Для QQueue нужно обязательно.

Нет, это значит, что методы можно повторно вызывать.

А как синхронизировать? Приведите, пожалуйста, кусочки кода. Или сслыку.

Поиск же по форуму... :)
http://www.prog.org.ru/index.php?topic=14426.msg95463#msg95463

Все-таки стандартные средства (напр контейнер защищенный QReadWriteLock) позволяют сэкономить немало времени. Может стоит сначала проверить "насколько же они медленны"

Тогда почему не lock-free? (т.е. без всяких контейнеров и мутексов).

C++ (Qt)
QAtomicPointer <СData> theHead(0), theTail(0);
QAtomicInt theCount(0);
 
void PutData( CData * data )   // CData - Ваша структура 
{
 data->mNext = 0;
 while (true) {
  CData * head = theHead;
  if (head && !theHead.testAndSetAcquire(head, 0)) continue;  // stop reading
 
  if (theCount++ == 0) {   // list is empty
    theHead = theTail = data;
    return;
  }
 
  while (true) {
    CData * tail = theTail;
    if (!tail || !theTail.testAndSetAcquire(tail, data)) continue;  // add data to list
    tail->mNext = data;
    break;
  }
  if (head) theHead = head;   // enable reading
  break;
 }
}
 
CData * GetData( void )
{
 while (theCount) {
  CData * cur = theHead;
  if (!cur) continue;   // wait for writer
  if (!theHead.testAndSetAcquire(cur, cur->mNext)) continue;
  --theCount;
  return cur;
 }
 return 0;
}
 

Впрочем синхронизация все равно нужна. А главное - это капитально "выносит моск", напр я совсем не уверен что написал правильно. Ну если конечно Вы не боитесь трудностей...  :)

Как-то вариант, предложенный Old проще для восприятия. Тоже недавно интересовался подобным вопросом.
Не совсем разобрался в эффективности применения QAtomic

Никакие атомарные операции тут не нужны. В 99.99% случаев вполне достаточно обычного мьютекса или крит. секции.

Передавать данные между потоками ненужно, лучше данные пошарить.
Создайте разделяемый ресурс.
Например пусть это будет класс.


Когда один поток обращается к данным, например для записи, обрамляете этот код вызовами
shareData->lock();
// синхронизируемый код
shareData->unlock();

Когда другой поток обращается к данным, например для чтения, также обрамляете код вызовами
shareData->lock();
// синхронизируемый код
shareData->unlock();


Если данных много и они должны обрабатываться постоянно, то можно сделать контейнер для хранения общих данных. Например вектор, а не char*, как в примере выше.
Тогда если !vector.isEmpty(), значит можно данные забирать и обрабатывать. Если нет, то ничего делать получающему потоку не нужно.
А передающий поток будет просто делать vector.push_back(item);

А в каких случаях целесообразно применять атомарные операции?

Когда хотите изобрести свою синхронизацию.

lock/unlock мьютекса это как раз атомарные операции.

А можно QSemaphore c фабрикой попробовать. разве нет?

В общем случае когда время выполнения "того что надо сделать под локом" достаточно мало. Напр то же добавление/извлечение из контейнера. Если такие операции интенсивны, то прямолинейное использование "честного" мутекса (который останавливает нитку) будет печально по скорости. Можно использовать нечестный (unfair), это просто

C++ (Qt)
struct CAtomMutex {
 void Lock( void )
 {
   while (!mVal.testAndSetAcquire(0, 1)) 
     QThread::yieldCurrentThread();
 }
 
 void Unlock( void )
 {
   mVal = 0;
 }
 
 QAtomicInt mVal;
};

Возможно и "никак", тупо ждем пока откроется мутекс. В этом есть смысл если по задаче мы знаем что время ожидания достаточно мало (ориентир < 0.3 sec)

Интересная статья по поводу QAtomic http://woboq.com/blog/introduction-to-lockfree-programming.html

Да что ж он, гад, делает! Дает код, показывает что в 10 раз быстрее, а потом (так, между прочим, в конце статьи) говорит что там баг, и как фиксить хз. Он же меня запутал!!! :)

Да, если интересно почему баг (там не очень ясно почему) - я поясню, это заморочка капитальная. А в общем ну его нафиг, тот lock-free, достигнутая чудесная скорость не окупает мозговых затрат на нее.

Зато повод узнать о существовании QAtomic и его возможностях  :)

Не путай людей своей бредятиной.
lock() и unlock() это атомарные операции.

Применение или не применение их никак не связано с временем выполнения кода, который между ними.

Целесообразность определяется не тем сколько времени выполняется код под локом, важна архитектура и потребности.

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

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

А что, если сделать просто:

C++ (Qt)
//Custom class - thread safe queue.
class SafeQueue<T> : QQueue {
    public:
        //Constuctor.
        SafeQueue() : Queue<T> {}
 
        //Enqueue
        void enqueue(T) {
            QMutexLocker locker(&mutex);
            QQueue::enqueue(T);
        }
 
        //Dequeue.
        T dequeue(T) {
            QMutexLocker locker(&mutex);
            return QQueue::dequeue();
        }
 
    private:
        QMutex mutex;
}

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

Правильно - да, быстро - не очень.

Edit: не совсем правильно :)  QQueue::dequeue() полагает что контейнер не пустой, надо добавить проверку

А как вы видите организацию поллинга для элементов вашей очереди? Как нить получатель будет определять, что в очереди появились данные?

Думаю поллинг лучше организовать так:

    1. Собрался лимит количества элементов - высылаю сигнал о готовности сразу. Это когда большой поток данных.

    2. Если лимит еще не собрался, но что то есть, то высылаю сигнал о готовности по таймауту. Это при непостоянном / медленном обмене данных.

Где сигнал о готовности - уведомление о наличии данных в очереди.

Накидайте это в коде (псевдокоде) и сравним с тем решением, ссылку на которое я вам привел.
Там такой сигнал как раз и посылается специальным объектом синхронизации QWaitCondition.

Тут может создаться впечатление что это какое-то "особое" решение. В действительности схема с
(Q)WaitCondition всегда одна и та же, другой просто нет. Конечно можно ее переписать, и она будет работать. А можно еще проще - сделать сигнал с QueuedConnection, и Qt за нас все сделает.

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

C++ (Qt)
//Custom class - thread safe queue.
class SafeQueue<T> : QQueue {
    public:
        //Constuctor.
        SafeQueue() : Queue<T>, mAbortFlag(false) {}
 
        void setAbort( void ) { mAbortFlag = true; }
 
        //Enqueue
        void enqueue( const T & data ) 
        {
            if (mAbortFlag) return;
            while (!mLock.testAndSetOrdered(0, 1)) {
             if (mAbortFlag) return;
             QThread::yieldCurrentThread()
            }
            QQueue::enqueue(data);
            mLock  = 0;
        }
 
        //Dequeue.
        T dequeue( void ) 
        {
            while (true) {
               if (mAbortFlag) return T(); 
               if (!mLock.testAndSetOrdered(0, 1)) 
                QThread::yieldCurrentThread();
               else {
                if (!size()) {
                 mLock = 0; 
                 QThread::yieldCurrentThread();
                 continue;
               } 
               T data = QQueue::dequeue();
               mLock = 0;
               return data; 
             }
            }
            return T();
        }
 
    private:
        bool mAbortFlag;
        QAtomicInt mLock;
}

Проще - возможно (как сказал бы Верес "сделайте тупо" :) ), а будет ли это решение сделанное Qt за нас - быстрее? Да и проще тоже сомнительно? :)

Вы бы не будили лиха - пока оно тихо. А то попрет "фонтан мысли"  :'(

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

Величина 0.3 секунды - это default параметр в OpenMP, типа "время прокрутки". Нитка усыпляется и перестает грузить проц по истечении этого интервала. Время совсем не маленькое

А что тут "велосипедного"? QAtomicInt, QWaitCondition и др. - штатные средства синхронизации. Нет готовых классов поюзать - ну что-то нужно и самому написать (чтоб совсем не облениться  :))

Предлагаю проверить на тестах, я возьму на себя реализацию с atomic

Какое OpenMP? Какие 0.3 секунды?!

Никаких классов писать не нужно, все уже давно написано. В 99.9999999999% случаев существующих средств синхронизации более чем достаточно.
Есть std, хотя мне не очень оно нравится, есть Qt'шные обертки, есть boost в конце концов.
Мьютексы, разные локи, wait-notify средства синхронизации и межпоточного взаимодействия.

Нужно просто понимать принцип их работы и уметь ими пользоваться.

Для меня штатной в Qt является система сигнал/слот. Всё остальное и есть велосипедостроение. Впрочем, это вопрос терминологии.

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

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

2xokc Вы бы подробные рассказали про архитектуру вашего сервера. И откуда берётся боспокойство о масштабируемости.

Ой, да долго рассказывать. Попробую.
По инфинибенду 10 раз в секунду блоками по 288*54 кБайт независимо друг от друга поступают в отдельных потоках данные с суммарной интенсивностью до 1 ГБайта в сек. Дальше с помощью CUDA оно асинхронно сплитится на 54*4 блока и по кольцу записывается на рейд из 8 SSD дисков, а так же кешируется в кольцевом буфере в оперативке. В этой части Qt практически не используется.
Дальше появляются пулы потоков с ридерами (до 1024 шт.), которые читают эти данные с разной скоростью в зависимости от нагрузки на сервер (от 10 р/с до 80 р/с). Дальше эти ридеры производят над считанными данными разные преобразования, результаты которых оправляются в сетку. Так вот кусок с ридерами сделан на Qt и сейчас взаимодействие между пулами потоков ридеров и сетевой подсистемой производится через сигнал-слот.
Сейчас оно крутится на 2-х процессорном Windows сервере на базе 12-ти ядерных Xeon. И периодически (как обычно, труднопрогнозируемо) "впадает в кому" в момент отправки сигнала с QByteArray. Не факт, что это связано с узким горлом в Qt сигналах - система только что собрана полностью и до этого места руки ещё не дошли. Проблема также в том, что параллельно всему этому в отдельном процессе работает своя достаточно ресурсоемкая задача (там и CUDA и OpenMP), которая тоже не прибавляет здоровья системе. В итоге "в среднем по больнице" при 256 ридерах вижу общую загрузку процессоров на уровне 30-40%, что пока устраивает.
Беспокоит то, что в перспективе ожидается кратное увеличение количества пишущих потоков (причем, возможно, существенное - правда при сохранении общей интенсивности потока данных за счет уменьшения количества блоков). Ну и ридеров может стать побольше. Поэтому и переживания о том, не упремся ли мы потом в производительность сигнал/слотов.

Я правильно понял, что потоки ридеры после обработки данных, передают их потоку сендеру, который и отправляет их в сеть? И для этой передачи используются сигналы?

В целом - да. Но сетевой поток в общем случае тоже не один - там пул потоков (клиентов может быть до 256 шт).

Посылка сигнала между потоками сводится к помещению события в Qt-шную очередь сообщений. Вам без очереди все равно не обойтись, поэтому пока не вижу серьезных причин отказываться от Qt-шной. В будущем, при необходимости, вы всегда сможете сделать свою специальную очередь для этого. Но без дополнительных тестов, показывающих что тормозит именно она, я бы пока не дергался.
А вот как раскидываются задачи на отправку между потоками сендерами? Или у вас по потоку на клиента и всем передаются одинаковые данные?
И как вы определили, что "в кому" впадает именно при отправке сигнала?

/* И выбор Qt для решения этой задачи удивляет. Не очень понятно для чего он здесь. Ну это риторический вопрос. :) */

С этим-то всё ясно, об этом-то сейчас и думаю. В том числе и в этом топике :)

Нет, на клиента не по потоку. Сетевые клиенты собраны в пулы, каждый пул обрабатывает один поток. Количество таких пулов пока задается вручную при старте сервера. Трафик, постоянно и равномерно отправляемый на клиента, составляет: в худшем случае 1 МБ/с, в среднем -  256 кБ/с.

Периодически удаётся "попасть" на это место при запуске сервера из-под отладчика. Последнее разумное место в стеке до AV это как раз вызов сигнала. Но, повторюсь, вопрос до конца не исследован, может это не причина, а следствие.

Не менее риторический ответ - по историческим причинам :). Фоново уже думаю о том, чтобы уйти в что-нибудь менее тяжёлое типа libuv или asio. Может ещё ZeroMQ. Но, там много всяких плюшек от Qt используется типа соединения с БД, логирование, сеть, QSettings, плагины, QSingleApplication и т.п. Не хотелось бы всё это заменять на набор разнородных костылей. В любом случае, делать это только ради того, чтобы гордо сказать, что у "меня на сервере нет Qt" не будем :).
А вот "со своей специальной очередью" - думаем.

Развил здесь (http://www.prog.org.ru/index.php?topic=27837.msg202663#msg202663)
Мда, запас прочности Qt намного больше чем я предполагал  :)

Все равно не очень понятно. Все клиенты получают одинаковые данные. А отправкой данных занимается одна нить на несколько клиентов?

Все клиенты получают разные данные. Вернее могут получать и одинаковые, но это частный случай. Ридеры читают разные места из кольцевого буфера - клиентам уходят разные данные. И - да, отправкой данных занимается одна нить на несколько клиентов. Считайте, что это большой архив системы видеонаблюдения. Клиент "смотрит" тот участок видео, какой ему понадобится, и с той скоростью, какую может позволить себе сервер или с какой может "справиться" сам клиент.

Кстати, таки определили из-за чего оно валится. У нас в приложении в каждом из потоков-читателей открыт один и тот-же файл, из которого и производится весьма интенсивное чтение блоками по 188 кБайт через QFile::read. Так вот, если QFile открыт без опции QIODevice::Unbuffered, то на уровне Qt производится кеширование, снимающее проблему низкой производительности при чтении из файла мелким порциями. И в нашем случае оно работает криво, т.е. приводит к регулярному AV. Так что проблема решилась добавлением к флагам при открытии файла QIODevice::Unbuffered.

Так а что значит: в вашем случае работало кривой?
В чем кривизна?

Вываливалось в access violation внутри QFile::read. Подебажил исходники - внутри read если не включен QIODevice::Unbuffered производится кеширование через внутренний QByteArray. Дальше разбираться не стал - добавил QIODevice::Unbuffered в QFile::open и всё заработало нормально. В нашей задаче даже кеш ОС отключить стоило-бы, так что лишние "оптимизации" тут со стороны Qt нам ни к чему.