Параллельный поток

[Eugene K]

Есть поток, где идет обработка данных параллельно основному потоку.

При обработке данных должна быть возможность нажимать на кнопки в программе.

Код:

//класс параллельного потока обработки данных
class MyThread : public QThread, public Ui::MainWindow
{
    Q_OBJECT

public: 
	volatile bool stopped; 
	MyThread(){	stopped = false; };
	~MyThread() {};
	char flag1, flag2; 
    
      void run()
	{
	while (!stopped) 
	{
	mutex.lock();
	if(flag2==1) { flag2=0; emit s21();}
	mutex.unlock();	
           }
	}

	void MyThread::stop() 
	{ 
	stopped = true; 
	}
      
signals:
	void s21(void);
	
private:
	QMutex mutex;	


//конструктор главной формы
frmMain::frmMain() : QWidget(0, Qt::Window)
{
	connect(pushButton_8, SIGNAL(clicked()), SLOT(an_t0()));
	connect(&thread1, SIGNAL(s21()), SLOT(an_t21()));
}


void frmMain::an_t0()
{
  
    if(lst.at(0).toInt()==21) 
    {
    thread1.start(QThread::LowPriority);
    thread1.flag2=1;         
    }


void frmMain::an_t21()
{
	//обработка данных

При работе программы параллельный поток обработки данных замораживает окно программы и кнопки невозможно нажимать.

Где я допустил неточность?

[Авварон]

Где я допустил неточность?

Грубо говоря - везде. Чуть точнее: an_t21() вызывается в _главном_ потоке.
Еще чуть точнее - выкиньте всё нафиг, опишите что вам надо словами сюда и вам скажут как делать правильно.

ЗЫ: никогда, никогда не наследуйте QThread для реализации бизнес-логики, тем более не наследуйте от нее окно.

[Eugene K]

Если словами, то на Qt4.1 (Windows) мне надо организовать четыре параллельных потока: первый принимает данные по сети, второй - обрабатывает эти данные, третий - вычисляет и фильтрует обработанные данные, четвертый - передает отфильтрованные данные по сети дальше.

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

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

[qate]

Qt4.1 вышло в 2012 г. - какая причина использовать эту версию ?

> мне надо организовать четыре параллельных потока: первый принимает данные по сети, второй - обрабатывает эти данные, третий - вычисляет и фильтрует обработанные данные, четвертый - передает отфильтрованные данные по сети дальше.

прием и передачу не надо пихать в отдельные потоки - работай по событиям
для второго и третьего используй https://doc.qt.io/qt-5/qtconcurrentrun.html
если так не пойдет - опиши почему ?

[Авварон]

Если словами, то на Qt4.1 (Windows) мне надо организовать четыре параллельных потока: первый принимает данные по сети, второй - обрабатывает эти данные, третий - вычисляет и фильтрует обработанные данные, четвертый - передает отфильтрованные данные по сети дальше.

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

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

Откуда требование в 4 потока?
Начните с однопоточного варианта, но с задумкой распараллелить, как сказано выше через QtConcurrent.
Начните с воркера-приемщика, положите туда сокет, сделайте там начальную валидацию данных, когда пришел чанк, достаточный для обработки, вызывайте функцию обработки.
Функция обработки должна быть свободной (или статической функцией), но отдельной от класса-приемщика, принимающая данные и возвращающая обработанные данные. Так ее можно будет легко завернуть в QtConcurrent.
Аналогично, фильтрация - свободная функция, принимает выход от обработчика.
Дальше, сделайте воркер-отправщик, принимающий слотом фильтрованные данные.

Теперь как это все подружить - сперва сделайте всё в главном потоке. Вероятно, будет тормозить обработка - сперва проверьте, что это так (либо через профилировщик, либо QElapsedTimer вставьте). Если тормозит обработка, выносите ее в поток(и) через QtConcurrent - я бы на обработку+фильтрацию делал одну задачу, но, в зависимости от объема данных, можно использовать и MapReduce. Делать отдельно обработку и фильтрацию не надо - у вас и так будет параллельно может исполнятся несколько задач обработки+фильтраци.и Результат от задачи получайте через QFutureWatcher  в том же главном потоке.
Если тормозит прием по сети (хотя вряд ли), можно завести объект QThread (не наследоваться!) и переместить приемщик, отправщик и FutureWatcher туда. Если у вас всё правильно разделено в однопоточном варианте, то проблем добавить тред не возникнет - это дело 10 строк (start, 3*moveToThread, 3*connect(&QThread::finished), quit, stop)

[Igors]

А не лучше ли задействовать QThreadPool скармливая ему возникающие задачи?

[Авварон]

А не лучше ли задействовать QThreadPool скармливая ему возникающие задачи?

конкаррент именно это и делает

[Igors]

конкаррент именно это и делает

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

[Авварон]

конкаррент именно это и делает

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

QtConcorrent::run ничего не распределяет и не вычисляет. Он лишь избавляет от секса с голыми указателями, мьютексами и наследованием раннаблов.

[Igors]

QtConcorrent::run ничего не распределяет и не вычисляет.

Это не так, он задумывался как аналог tbb или OpenMP с ключом schedule dynamic. Сначала он дает нитке одну задачу и после завершения анализирует время выполнения. Потом нитка может получить уже 2 (или 10) задач или опять одну, там довольно длинная песня, смысл - оптимально распределить нагрузку.

[Авварон]

Шта? Вы где это в коде увидели?

[Igors]

Шта? Вы где это в коде увидели?

Вот "видимая часть" айсберга, см также timeBefore/timeAfter

Код

C++ (Qt)
   ThreadFunctionResult forThreadFunction()
    {
        BlockSizeManagerV2 blockSizeManager(iterationCount);
        ResultReporter<T> resultReporter(this);
 
        for(;;) {
            if (this->isCanceled())
                break;
 
            const int currentBlockSize = blockSizeManager.blockSize();
 
            if (currentIndex.load() >= iterationCount)
                break;
 
            // Atomically reserve a block of iterationCount for this thread.
            const int beginIndex = currentIndex.fetchAndAddRelease(currentBlockSize);
            const int endIndex = qMin(beginIndex + currentBlockSize, iterationCount);
 
            if (beginIndex >= endIndex) {
                // No more work
                break;
            }
 
            this->waitForResume(); // (only waits if the qfuture is paused.)
 
            if (shouldStartThread())
                this->startThread();
 
            const int finalBlockSize = endIndex - beginIndex; // block size adjusted for possible end-of-range
            resultReporter.reserveSpace(finalBlockSize);
 
            // Call user code with the current iteration range.
            blockSizeManager.timeBeforeUser();
            const bool resultsAvailable = this->runIterations(begin, beginIndex, endIndex, resultReporter.getPointer());
            blockSizeManager.timeAfterUser();
 
            if (resultsAvailable)
                resultReporter.reportResults(beginIndex);
 
            // Report progress if progress reporting enabled.
            if (progressReportingEnabled) {
                completed.fetchAndAddAcquire(finalBlockSize);
                this->setProgressValue(this->completed.load());
            }
 
            if (this->shouldThrottleThread())
                return ThrottleThread;
        }
        return ThreadFinished;
    }

[Авварон]

Шта? Вы где это в коде увидели?

Вот "видимая часть" айсберга, см также timeBefore/timeAfter

Отлично, где в QtConcurrent::run это используется?
Это используется в QtConcurrent::map, ну да и бог с ним, я же написал ровно что и вы - сперва попробовать следать через run() который тот же раннабл

[Eugene K]

Пробую использовать moveToThread(thread)

но при создании объекта thread=new QThread() компилятор ругается:
cannot allocate an object of type 'QThread'
because the following virtual functions are abstract: error: virtual void QThread::run()

[Авварон]

Твою ж, у вас 4.1  Точно никак поновее нельзя взять?

Ладно, делать нечего - надо отнаследоваться от треда и в своем run() тупо вызвать exec(), как сделано в более новых версиях