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

[Igors]

любая нитка (кроме главной)

и главная тоже, если она не вызывает exec(), то и у неё не будет цикла обработки событий

Ну главная нитка не QThread и ее нельзя создать/удалить, но в принципе правильно: без QApplication::exec ничего не получится

[labview]

Привет!

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

И так задумка (чтобы не читать весь топик).

Eсть два потока. Для начала один записывает один считывает данные. Данные между потоками передаются не через сигналы и слоты, а через Qlist с использованием QWaitCondition. Должно по-моему выпоняться так, что стартуются два потока, второй ждёт кондиции и спит в это время. Первый при старте записывает в Qlist данные и будит второй поток. Тот считывает эти данные, когда его разбудили.

Систему QList+QWaitCondition я назвал Tasking (т.к. логически передаются задания с параметрами из одного потока во второй).

И так Tasking.h

Код:

#ifndef TASKING_H
#define TASKING_H

#include <QList>
#include <QWaitCondition>
#include <QMutex>
#include <QString>

struct task
{
    int tasknum;
    QString data;
};


class Tasking
{
private:
    QList<task> *taskList;
    QWaitCondition *cond;
    QMutex *mutex;
public:
    Tasking();
    ~Tasking();
    task getTask();
    void addTask(task newTask);
};

#endif // TASKING_H

Tasking.cpp

Код:

#include "Tasking.h"

Tasking::Tasking()
{
    mutex = new QMutex;
    taskList = new QList<task>;
    cond = new QWaitCondition;
}

Tasking::~Tasking()
{

}

task Tasking::getTask()
{
    if(!(taskList->count()==0))
    {
        mutex->lock();
        cond->wait(mutex, ULONG_MAX);
        mutex->unlock();
    }

    return taskList->last();
}

void Tasking::addTask(task newTask)
{
    cond->wakeOne();
    taskList->push_back(newTask);
}

Thread1 (который посылает задание второму)

Код:

#include "thread1.h"


Thread1::Thread1(Tasking *allTasks[], int tasksCount)
{
    Tasks = new Tasking * [tasksCount];
    for(int i = 0; i < tasksCount; i++)
    Tasks[i] = allTasks[i];
}

void Thread1::run()
{
    task myTask;
    myTask.tasknum = 1;
    myTask.data = "something";
    Tasks[0]->addTask(myTask);
    emit showData(myTask.data);
}

Thread2 (который ждёт задания)

Код:

#include "thread2.h"

Thread2::Thread2(Tasking *allTasks[], int tasksCount)
{
    Tasks = new Tasking * [tasksCount];
    for(int i = 0; i < tasksCount; i++)
    Tasks[i] = allTasks[i];
}

void Thread2::run()
{
    task myTask;
    myTask = Tasks[0]->getTask();
    emit showData(myTask.data);
}

Main

Код:

#include <QtGui/QApplication>

#include "mainwidget.h"
#include "Tasking.h"
#include "thread1.h"
#include "thread2.h"

int main(int argc, char *argv[])
{
    QApplication a(argc, argv);
    MainWidget w;
    w.show();

    Tasking *allTasks[1];

    Tasking *tasking1 = new Tasking;
    allTasks[0] = tasking1;
    Thread1 myThread1(allTasks, sizeof(allTasks)/sizeof(allTasks[0]));
    Thread2 myThread2(allTasks, sizeof(allTasks)/sizeof(allTasks[0]));

    QObject::connect(&myThread1, SIGNAL(showData(QString)), &w, SLOT(addText(QString)));
    QObject::connect(&myThread2, SIGNAL(showData(QString)), &w, SLOT(addText(QString)));

    myThread1.start();
    myThread2.start();
    myThread1.wait();
    myThread2.wait();

    return a.exec();
}

Спасибо!

[Igors]

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

Да много чего 

1)

Код:

 if(!(taskList->count()==0))
    {
        mutex->lock();
        cond->wait(mutex, ULONG_MAX);
        mutex->unlock();
    }

То есть если список НЕ пустой. то ждем? Отрицание надо убрать

2) Если QList используется 2 или более нитками и есть хотя бы 1 операция записи в него, то все обращения к такому QList (как по записи так и по чтению) должны быть защищены (напр мутексом) как уже обсуждали в этой теме

3) Если Вы хотите делать свой цикл нитки - делайте, напр так

Код

C++ (Qt)
void Thread2::run( void )
{
   while (!mStop) {
     ...
   }
}
 

А то у Вас в run по 1 задаче стоит

4)

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

Во-первых, в коде наоборот, сначала будит, потом записывает

Код:

void Tasking::addTask(task newTask)
{
    cond->wakeOne();
    taskList->push_back(newTask);
}

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

Если Вы хотите класть задания в очередь и вынимать из нее, то есть простой и хороший класс QSemaphore с методами acquire и release. Разница в том что не нужно заморачиваться с выяснением "а спит ли уже нитка?". Добавили в очередь и вызвали release семафора. Если нитка 2 стоит на acquire, она разблокируется. А если нет - не беда, тогда не будет ждать на acquire когда до этого дойдет дело.

[BRE]

Набросаю свои мысли.

Код

C++ (Qt)
class Tasking
{
private:
	// Не знаю, для чего использовать здесь указатели, поэтому предлагаю сделать проще:
	QList<task> taskList;
	QMutex mutex;
	QWaitCondition cond;
 
public:
	task getTask();
	void addTask(task newTask);
};
 

Код

C++ (Qt)
task Tasking::getTask()
{
	QMutexLocker locker( &mutex );
	while( taskList.empty() )
		cond.wait( &mutex );
 
	return taskList.takeFirst();
}
 
void Tasking::addTask(task newTask)
{
	QMutexLocker locker( &mutex );
	taskList.append( newTask );
	cond.wakeOne();
}
 

[Igors]

Код

C++ (Qt)
task Tasking::getTask()
{
	QMutexLocker locker( &mutex );
	while( taskList.empty() )
		cond.wait( &mutex );
 
	return taskList.takeFirst();
}
 
void Tasking::addTask(task newTask)
{
	QMutexLocker locker( &mutex );
	taskList.append( newTask );
	cond.wakeOne();
}
 

Если getTask захватила мутекс и лист пуст - имеем dead lock. С семафором получается прекрасно

Код

C++ (Qt)
task Tasking::getTask()
{
	semaphore.acquire();
	QMutexLocker locker( &mutex );
	return taskList.takeFirst();
}
 
void Tasking::addTask(task newTask)
{
	QMutexLocker locker( &mutex );
	taskList.append( newTask );
	semaphore.release();
}
 

[BRE]

Если getTask захватила мутекс и лист пуст - имеем dead lock.

Да ну. А может стоит разобраться с работой wait condition. 

[labview]

Спасибо ребята!!! Подправил ошибки, работает!

На счёт цикла в потоке я знаю конечно, пока что делал для примера без цикла. Гляньте плиз на счёт мутекса, думаю теперь всё в порядке, или?

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

Код:

    if(taskList.count()==0)
    {
        cond.wait(&mutex, ULONG_MAX);
    }

Код:

#include "Tasking.h"

Tasking::Tasking()
{
}

Tasking::~Tasking()
{

}

task Tasking::getTask()
{
    if(taskList.count()==0)
    {
        cond.wait(&mutex, ULONG_MAX);
    }

    mutex.lock();
    task currentTask = taskList.last();
    mutex.unlock();

    return currentTask;
}

void Tasking::addTask(task newTask)
{
    mutex.lock();
    taskList.push_back(newTask);
    mutex.unlock();
    cond.wakeOne();
}

[BRE]

Хм.
Я не зря написал петлю, она нужна, так как в некоторых случаях поток может проснутся не вовремя (1).
Причем, этот код должен быть уже залочен мьютексом (2).

Код

C++ (Qt)
        QMutexLocker locker( &mutex );		// <<<<<<<<<<< 2
        while( taskList.empty() )				// <<<<<<<<<<< 1
                cond.wait( &mutex );
 
        return taskList.takeFirst();
 

[Igors]

Гляньте плиз на счёт мутекса, думаю теперь всё в порядке, или?

"Или"    Мутекс должен быть захвачен до вызова wait, поэтому правильно так

Код:

task Tasking::getTask()
{
    mutex.lock();
    if(taskList.count()==0)
        cond.wait(&mutex, ULONG_MAX);

    task currentTask = taskList.last();
    mutex.unlock();

    return currentTask;
}

wait освободит мутекс перед остановом нитки, а когда нитка снимется с ожидание - снова захватит

[BRE]

wait освободит мутекс перед остановом нитки, а когда нитка снимется с ожидание - снова захватит

Точно. 

Только цикл еще нужен, проверяющий count() и выполняющий wait(). 

[Igors]

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

Это если Вы берете задание и сразу удаляете его из очереди (как в варианте BRE). А если не удаляете (как у Вас) то "лист пустой или нет" ни о чем не говорит

Только цикл еще нужен, проверяющий count() и выполняющий wait(). 

Нет, цикл не нужен (хотя он ничего и не портит)

[BRE]

Нет, цикл не нужен (хотя он ничего и не портит)

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

[Igors]

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

Может Вас смущает это (QWaitCondition::wait)

Код

C++ (Qt)
mutax->unlock();
 
bool returnValue = d->wait(time);
 

Да, др. нитка может захватить освобожденный мутекс, но она не прорвется на wakeOne. т.к. будет остановлена на другом, внутреннем мутексе QWaitCondition (d->mutex). То есть pthread_cond_signal придет после pthread_cond_wait

[BRE]

Может Вас смущает это (QWaitCondition::wait)

Нет, меня смущает это:

Код

C++ (Qt)
    if(taskList.count()==0)
        cond.wait(&mutex, ULONG_MAX);
 

вместо этого:

Код

C++ (Qt)
    while( taskList.count()==0 )
        cond.wait(&mutex, ULONG_MAX);
 

И я писал выше почему.

[Igors]

И я писал выше почему.

Выше я вижу

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

Весьма мутное/общее "почему"  Поясните что то за  "не системные ошибки", "некоторые потоки" и "моменты времени"