Очереди, самодельные сигналы

[labview]

Привет!

Я, как программист на LabVIEW предпочитаю многопоточное программирование. Этот язык программирования отличается тем, что в нём автоматически создаются параллельные потоки, если нет прямой связи между функциями. То есть защищаться нужно от так называемых Race Conditions, а не бороться с трудностями параллельного программирования циклов. Короче наоборот, чем в C++.

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

Спасибо!

[labview]

Самодельные сигналы можно делать с помощью emit, я правильно понимаю?

Вопрос с очередями (или FIFO) всё же остаётся.

[SimpleSunny]

Да, правильно, свои сигналы "запускают" по emit.

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

Код

C++ (Qt)
class storage
{
    QReadWriteLock lock;
    QList <int> list;
public:
    int getLast() {lock.lockForRead();i = list.last(); lock.unlock(); return i;}
    void addLast(int i) {lock.lockForWrite(); list.push_back(i); lock.unlock();}
};

Или можно соединить потоки через "сигнал-слот" для обмена данными.

[labview]

QList видимо хорошая штука, прочтал немного про него, а так же нашёл QQueue. Видимо как раз то, что мне нужно.
Так же уже нарыл инфу о Mutex.

Буду пробовать, спасибо!

[Igors]

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

Код

C++ (Qt)
class storage
{
    QReadWriteLock lock;
    QList <int> list;
public:
    int getLast() {lock.lockForRead();i = list.last(); lock.unlock(); return i;}
    void addLast(int i) {lock.lockForWrite(); list.push_back(i); lock.unlock();}
};

Все верно, но, к сожалению, при интенсивном чтении такого QList 4 нитки работают медленнее чем 1 

[SimpleSunny]

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

[labview]

Спасибо за ответы.

Меня собственно интересуют следующие функции/возможности межпоточного синхронизированного обмена информацией без потерь данных:

1. Создание FIFO, где можно указать тип элемента и максимальное количество элементов, с поддержкой безграничного количества элементов.

2. Запись элемента в конец FIFO из любого из потоков. Если FIFO переполнено, то поток ждёт (застревает без пожирания CPU) пока в FIFO не появится место для элемента.

3. Предпросмотр первого элемента FIFO, без вытаскивания его из FIFO.

4. Доступ к информации о количестве элементов в FIFO, а так же ко всем элементам в FIFO в виде массива, без вытаскивания их из FIFO.

5. Закрытие FIFO, при котором все элементы уничтожаются, референс на FIFO становится недействительным. Если какой то поток находится в ждущем режиме IDLE (хочет считать элемент), то функция считывания прекращает своё действие и возвращает ошибку, т.к. референс на FIFO более недействителен.

6. Запись элемента в конец FIFO, при этом отличие от пункта 2, то, что если FIFO переполнено, то первый элемент удаляется.

7. Запись элемента в начало FIFO (то есть это уже будет LIFO).

8. Чтение первого элемента FIFO. При этом, если в FIFO нет элементов, то эта функция ждёт (находится в режиме IDLE, без пожирания CPU) пока элемент не появится. Так же здесь должен быть параметр Timeout, который говорит о том, как долго функция должна оставаться в состоянии IDLE. По прохождении времени Timeout, функция возвращает ошибку.

9. Удаление всех элементов из FIFO, без уничтожения самой FIFO. Т.е. референс остаётся действительным. Эта функция возвращает все уничтоженные элементы очереди в виде массива.


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

Ещё раз огромное спасибо.

[Igors]

А за счет чего, в данном случае, получается такие тормоза?

QMutex, QReadWriteLock используют ОС реализацию mutex, которая может работать очень медленно при наличии интенсивной конкуренции за ресурс. Для примера OpenMP имеет настраиваемый параметр - время прокрутки (холостого хода) нитки перед тем как ее усыпить. По умолчанию это значение 200 ms - то есть 0.2 секунды. Делайте выводы

[Igors]

Меня собственно интересуют следующие функции/возможности межпоточного синхронизированного обмена информацией без потерь данных:

Если говорить о структуре данных, то для всего что Вы перечислили ниже, QList прекрасно подходит т.к. он оптимизирован для вставок/удаления элементов в голову и в хвост. Также важно при вставке/удалении в QList адрес других элементов не меняется.

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

[lit-uriy]

одно предостережение, у QList есть ограничение на размер (ограничен индексом типа int), он не безграничен

[labview]

Привет!

Пытаюсь разобраться. Немного не понимаю как в контрукторе нити добавить другие параметры кроме name. Может кто нибудь помочь?
Ну скажем например какое нибудь число n.

Спасибо, код с комментариями прилагаю.

Код:

#include <iostream>
#include <QThread>
#include <QList>
#include <QReadWriteLock>

//Дефинируем класс состоящий из FIFO + lock для межпоточного обмена данными
class MyQueue
{
    QReadWriteLock lock;
    QList <int> list;
public:
    int i;
    int getLast() {lock.lockForRead();i = list.last(); lock.unlock(); return i;}
    void addLast(int i) {lock.lockForWrite(); list.push_back(i); lock.unlock();}
};

//Дефинируем класс MyThread полученный из QThread
class MyThread : public QThread {
public:
  MyThread( std::string a = "MyThread" );
  MyQueue myQueue; //внутри нити создаём нашу очередь
private:
  std::string name;
  virtual void run();
};

//Как я понимаю это конструктор
MyThread::MyThread( std::string a ) : name(a)
{}

//Перезагружаем функцию run() тем, что нить должна делать после запуска
void MyThread::run()
{
    int  b = 1;
    myQueue.addLast(b);
    b = myQueue.getLast();
    std::cout << name <<  b  << std::endl;
}


int main()
{
  MyThread a("A");
  MyThread b("B");
  a.start();
  b.start();
  a.wait();
  b.wait();
}

[kuzulis]

Как вариант. В конструкторе потока должен быть добавлен указатель на объект типа MyQueue.
Тогда несколько потоков смогут иметь доступ к MyQueue.

Код:

class MyThread : public QThread {

    Q_OBJECT

public:
  MyThread(MyQueue *queue, QObject *parent = 0);
private:
  MyQueue *myQueue; 
protected:
    void run();
};

Код:

MyThread::MyThread(MyQueue *queue, QObject *parent)
        : QThread(parent), myQueue(queue)
{
}

Код:

void MyThread::run()
{
    int  b = 1;
    myQueue->addLast(b);
    b = myQueue->getLast();
    std::cout <<  b << std::endl;
}

Код:

int main()
{
  MyQueue mq;
  MyThread a(&mq);
  MyThread b(&mq);
  a.start();
  b.start();
  a.wait();
  b.wait();
}

[SimpleSunny]

Ваш код содержит потенциальную ошибку.

Код

C++ (Qt)
int i;
int getLast() {lock.lockForRead();i = list.last(); lock.unlock(); return i;}
void addLast()...

Допустим есть 3 потока: 2 читают (1й и 2й), 1 (3й) пишет.
1й вызывает getLast() и перед return i переходит переключение на 3й поток, тот добавляет какое-то значение в конец, переключение на 2й поток, который успевает сделать i = list.last(), переключение на 1й поток. Дальше все зависит от положение солнца и влажности воздуха.

Как добавлять параметры в конструкторе показали, в принципе, выше.
Можно также сделать отдельный метод для передачи параметров (setParameter(int)), если параметров много, можно завернуть их в отдельную структуру \ клас.

[labview]

Дело в том, что у каждого потока будет иметься своя собственная очередь. То есть два читателя одной очереди исключаются.

[SimpleSunny]

Дело в том, что у каждого потока будет иметься своя собственная очередь. То есть два читателя одной очереди исключаются.

Если у каждого потока своя очередь, тогда зачем междупотоковая синхронизация (QReadWriteLock)?