Два потока работают с QQueue

[Yegor]

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

Код

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;
}

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

[Igors]

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

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

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

[Old]

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

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

[Yegor]

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

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

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

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

[Old]

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

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

[Igors]

Накидайте это в коде (псевдокоде) и сравним с тем решением, ссылку на которое я вам привел.
Там такой сигнал как раз и посылается специальным объектом синхронизации 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;
}

[Old]

А можно еще проще - сделать сигнал с QueuedConnection, и Qt за нас все сделает.

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

[Igors]

как сказал бы Верес "сделайте тупо"

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

[xokc]

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

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

[Igors]

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

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

..и начинать городить велосипедный огород?

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

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

[vulko]

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

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

..и начинать городить велосипедный огород?

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

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

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

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

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

[xokc]

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

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

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

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

В 99.9999999999% случаев существующих средств синхронизации более чем достаточно.

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

[Old]

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

[xokc]

Ой, да долго рассказывать. Попробую.
По инфинибенду 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%, что пока устраивает.
Беспокоит то, что в перспективе ожидается кратное увеличение количества пишущих потоков (причем, возможно, существенное - правда при сохранении общей интенсивности потока данных за счет уменьшения количества блоков). Ну и ридеров может стать побольше. Поэтому и переживания о том, не упремся ли мы потом в производительность сигнал/слотов.

[Old]

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