QThread - parallel

[sgrant327]

Hi,

I am trying to get an application working that utilizes 4 instances of the same thread. I have another thread called ThreadManager that hands data to each of the 4 threads...

I am processing folders that contain 4000+ files. I must read and update data in each file. They can all be done simultaneously...without depending on each other.

My 4 workers each process a file...with the Manager controlling which file is next to be processed and handing it off to the next thread that is waiting...

Now for the problem. I want to update the GUI as each file is processed. I have signals coming out of each worker that is attached to a unique slot in the Manager. The Manager's slots emit signals that are connected to the main app. No problem except that the GUI lags behind...quite a bit...when using QueuedConnections. I currently have DirectConnections for the worker->manager singals and QueuedConnections from the manager->main. If I use DirectConnection for the manager->main, the app crashes with the first worker signal emitted.

On another note, when I dispatch a file to a worker, if I DO NOT use a wait() on that worker, the app tends to crash after a few iterations. The only way I seem to be able to get through an entire directory is by calling wait(ULONG_MAX)...any ideas?

Thanks,
Sam

[borisbn]

1. you can not use DirectConnection from thread to gui
2. is it very important to show to user all of that 4000+ filenames ? Show him just a done percent.
3. each of your thread must check if user wants to close application, and do break if it is.
4. if lines above didn't help, show code of your gui, manager and workers without actually processing algorythm (just a comment like "here is actually processing code") to decrease your message in post

[wysota]

No problem except that the GUI lags behind...quite a bit...

If you have less than 5 processing units in your machine then the GUI thread will be getting less and less CPU time so it will effectively be working slower.

On another note, when I dispatch a file to a worker, if I DO NOT use a wait() on that worker, the app tends to crash after a few iterations. The only way I seem to be able to get through an entire directory is by calling wait(ULONG_MAX)...any ideas?

Can we see the exact code?

By the way, did you think about using QtConcurrent (QRunnable at least) instead of manually spawned threads?

[sgrant327]

Wysota,

The machine I am working on has dual quad-Xeon processors, essentially 8 cores.

I am unable to give you the exact code, but here is a quick synopsis:

Main:

1. Create ThreadManager
2. Connect TM's signals to main's GUI update slots
3. Pass 'default' settings to TM
4. Compile list of files in selected directory
5. Pass list to TM
6. TM.start();

ThreadManager:

1. Create WorkerThread1
2. Connect WT1's signals to TM's slots
3. Pass 'default' settings to WT1
4. Create WorkerThread2
5. Connect WT2's signalst to TM's slots
6. Pass 'default' settings to WT2

--start running:

while( !fileIDX == fileList.count() )
{
    if( !WT1.isRunning() )
   {
     WT1.filename = list[fileIDX];
     WT1.start();
     fileIDX++;
    }
    if( !WT2.isRunning() )
   {
     WT2.filename = list[fileIDX];
     WT2.start();
     fileIDX++;
    }
}

while( !fileIDX == fileList.count() )
{
    if( !WT1.isRunning() )
   {
     WT1.filename = list[fileIDX];
     WT1.start();
     fileIDX++;
    }
    if( !WT2.isRunning() )
   {
     WT2.filename = list[fileIDX];
     WT2.start();
     fileIDX++;
    }
}

To copy to clipboard, switch view to plain text mode 

Basically that is it. Each WT opens it's corresponding file, makes changes to it and saves.

--Sam

[spud]

Unless you have left out critical code, you do not seem to be ensuring thread safety. You need a QWaitCondition and a QMutex. Check out the mandelbrot example.
Here is the relevant code for a safe construction of a worker thread.

class JobThread : public QThread
{
public:
	JobThread(QObject *parent)
		: QThread(parent)
	 {
		abort = false;
		start(LowPriority);
	 }
 
	 ~JobThread()
	 {
		mutex.lock();
		abort = true;
		condition.wakeOne();
		mutex.unlock();
		wait();
	 }
 
	 void addJob(Job job)
	 {
		QMutexLocker locker(&mutex);
		this->jobs << job;
		condition.wakeOne();
	 }
 
	 void run()
	 {
		 QMutexLocker lock(&mutex);
		 while(!abort) {
			 if (!jobs.isEmpty())
			 {
				 condition.wait(&mutex);
				 continue;
			 }
			 Job job = jobs.takeFirst();
			 lock.unlock();
 
			 while (!abort) {
				 // do heavy computing
				 process(job);
				 emit progress(...);
			 }
			 lock.relock();
		 }
	 }
private:
	mutable bool abort;
	QMutex mutex;
	QWaitCondition condition;
	QList<Job> jobs;
};

class JobThread : public QThread
{
public:
	JobThread(QObject *parent)
		: QThread(parent)
	 {
		abort = false;
		start(LowPriority);
	 }

	 ~JobThread()
	 {
		mutex.lock();
		abort = true;
		condition.wakeOne();
		mutex.unlock();
		wait();
	 }

	 void addJob(Job job)
	 {
		QMutexLocker locker(&mutex);
		this->jobs << job;
		condition.wakeOne();
	 }

	 void run()
	 {
		 QMutexLocker lock(&mutex);
		 while(!abort) {
			 if (!jobs.isEmpty())
			 {
				 condition.wait(&mutex);
				 continue;
			 }
			 Job job = jobs.takeFirst();
			 lock.unlock();

			 while (!abort) {
				 // do heavy computing
				 process(job);
				 emit progress(...);
			 }
			 lock.relock();
		 }
	 }
private:
	mutable bool abort;
	QMutex mutex;
	QWaitCondition condition;
	QList<Job> jobs;
};

To copy to clipboard, switch view to plain text mode 

I would also consider using QtConcurrent.

[wysota]

It would be easier if you did it with a wait condition or using QtConcurrent. The code you posted is... well... error prone. For instance QThread::start() doesn't start a thread, as you would expect. It returns immediately and the thread will be started some time afterwards, so isRunning() might return false before the thread actually gets started.

What you want can basically be obtained via:

void doSomething(const QString &fileName){
  doSomethingWithTheFile(fileName);
}
//...
QStringList fileNames = buildFileList();
QtConcurrent::map(fileNames, doSomething);

void doSomething(const QString &fileName){
  doSomethingWithTheFile(fileName);
}
//...
QStringList fileNames = buildFileList();
QtConcurrent::map(fileNames, doSomething);

To copy to clipboard, switch view to plain text mode 

No mess and will use all the available cores in your machine.

[sgrant327]

By using QtConcurrent, will I be able to update the GUI? I could not find anything in the docs about the relationship between QtConcurrent and the Main Thread (GUI).

Thanks for all the help wysota!

[spud]

See QFutureWatcher.

[sgrant327]

OK,

This could be because I am a noob, but...

class packageWindow : public QMainWindow
{
     Q_OBJECT
 
public:
     void processFile(const QString &myFile);
};
 
QStringList myList;
for( int i = 0; i < numFiles; i++ )
{
 myList.append(allFiles[i]->text());
}
QtConcurrent::map(myList,processFile);

class packageWindow : public QMainWindow
{
     Q_OBJECT

public:
     void processFile(const QString &myFile);
};

QStringList myList;
for( int i = 0; i < numFiles; i++ )
{
 myList.append(allFiles[i]->text());
}
QtConcurrent::map(myList,processFile);

To copy to clipboard, switch view to plain text mode 

The QtConcurrent call generates an error:

argument of type `void(packageWindow::)(const QString&)` does not match `void(packageWindow::*)(const QString&)`

argument of type `void(packageWindow::)(const QString&)` does not match `void(packageWindow::*)(const QString&)`

To copy to clipboard, switch view to plain text mode 

What does THAT mean?

[wysota]

You can't use a non-static member function from another class. It either has to be static or can't be a member function. What is allowed and what is not is described in Qt Concurrent docs.

[sgrant327]

Got it...

Took me a while to get that EVERYTHING the QtConcurrent thread uses has to be static.

Just having a bit of trouble with QFutureWatcher and QtConcurrent. Sometime the app crashes after about 20 files, other times it runs fine, but 'locks' the computer as all 8 cores are running at 100%...

[spud]

You can use

QThreadPool::globalInstance->setMaxThreadCount()

QThreadPool::globalInstance->setMaxThreadCount()

To copy to clipboard, switch view to plain text mode 

to limit the number of threads.
And if your application still is crashing, you are probably still accessing a shared variable in an unsafe way. I'm thinking of myList and allFiles.
It sounds like QtConcurrent::mappedReduced could be a candidate for you.

[wysota]

What exactly does your thread method do? Is there a chance you could post the code here?

[Lesiok]

Got it...

Took me a while to get that EVERYTHING the QtConcurrent thread uses has to be static.

Just having a bit of trouble with QFutureWatcher and QtConcurrent. Sometime the app crashes after about 20 files, other times it runs fine, but 'locks' the computer as all 8 cores are running at 100%...

Because as is writen in documentation Programs written with QtConcurrent automatically adjust the number of threads used according to the number of processor cores available. So they consume 100% available cores.
Your first idea with ThreadManager and WorkerThread is good. You must only create slot in ThreadManager and connect to them signal QThread::finished(). In this slot start worker with next file.